Repository: apache/arrow-rs
Pull request #6992 (2025-01-17 21:46:14)

Expose min/max values for Decimal128/256 and improve docs
# Which issue does this PR close?


- Related to https://github.com/apache/datafusion/pull/14126

# Rationale for this change
 
We need to know the min/max values for `Decimal128` and `Decimal256` in Datafusion (see https://github.com/apache/datafusion/pull/14126 from @waynexia )

Currently only the min/max values are exposed for `Decimal128`

Also it turns out that the exposed values is different than what is currently used, which is confusing

# What changes are included in this PR?
1. Expose min/max values for both `Decimal128` and `Decimal256`
2. Add documentation and examples
3. Rename the `pub(crate)` constants to be consistent
4. Deprecate the old (unused) `pub` constants

![Screenshot 2025-01-17 at 4 37 39 PM](https://github.com/user-attachments/assets/78c310e8-0e24-4e9f-8348-ae010d998e6a)

# Are there any user-facing changes?
Yes, deprecated some constants and made others public

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please call them out.
-->



Patch:
diff --git a/.github/workflows/parquet.yml b/.github/workflows/parquet.yml
index 19503fde7..4c46fde19 100644
--- a/.github/workflows/parquet.yml
+++ b/.github/workflows/parquet.yml
@@ -97,6 +97,8 @@ jobs:
         run: cargo check -p parquet --no-default-features
       - name: Check compilation --no-default-features --features arrow
         run: cargo check -p parquet --no-default-features --features arrow
+      - name: Check compilation --no-default-features --features simdutf8
+        run: cargo check -p parquet --no-default-features --features simdutf8
       - name: Check compilation --no-default-features --all-features
         run: cargo check -p parquet --all-features
       - name: Check compilation --all-targets
diff --git a/README.md b/README.md
index 7a3dc1dd2..7be578278 100644
--- a/README.md
+++ b/README.md
@@ -53,7 +53,7 @@ as the [`parquet`] and [`parquet-derive`] crates.
 
 This crate releases every month. We release new major versions (with potentially
 breaking API changes) at most once a quarter, and release incremental minor
-versions in the intervening months. See [this ticket] for more details.
+versions in the intervening months. See [ticket #5368] for more details.
 
 To keep our maintenance burden down, we do regularly scheduled releases (major
 and minor) from the `main` branch. How we handle PRs with breaking API changes
@@ -72,7 +72,7 @@ Planned Release Schedule
 | Feb 2025         | `54.2.0` | Minor, NO breaking API changes             |
 | Mar 2025         | `55.0.0` | Major, potentially breaking API changes    |
 
-[this ticket]: https://github.com/apache/arrow-rs/issues/5368
+[ticket #5368]: https://github.com/apache/arrow-rs/issues/5368
 [semantic versioning]: https://semver.org/
 
 ### `object_store` crate
@@ -96,9 +96,9 @@ In general, use panics for bad states that are unreachable, unrecoverable or har
 For those caused by invalid user input, however, we prefer to report that invalidity
 gracefully as an error result instead of panicking. In general, invalid input should result
 in an `Error` as soon as possible. It _is_ ok for code paths after validation to assume
-validation has already occurred and panic if not. See [this ticket] for more nuances.
+validation has already occurred and panic if not. See [ticket #6737] for more nuances.
 
-[this ticket]: https://github.com/apache/arrow-rs/issues/6737
+[ticket #6737]: https://github.com/apache/arrow-rs/issues/6737
 
 ### Deprecation Guidelines
 
diff --git a/arrow-array/src/array/list_array.rs b/arrow-array/src/array/list_array.rs
index bed0bdf88..b53bbe7ab 100644
--- a/arrow-array/src/array/list_array.rs
+++ b/arrow-array/src/array/list_array.rs
@@ -55,7 +55,9 @@ impl OffsetSizeTrait for i64 {
 }
 
 /// An array of [variable length lists], similar to JSON arrays
-/// (e.g. `["A", "B", "C"]`).
+/// (e.g. `["A", "B", "C"]`). This struct specifically represents
+/// the [list layout]. Refer to [`GenericListViewArray`] for the
+/// [list-view layout].
 ///
 /// Lists are represented using `offsets` into a `values` child
 /// array. Offsets are stored in two adjacent entries of an
@@ -123,7 +125,10 @@ impl OffsetSizeTrait for i64 {
 /// ```
 ///
 /// [`StringArray`]: crate::array::StringArray
+/// [`GenericListViewArray`]: crate::array::GenericListViewArray
 /// [variable length lists]: https://arrow.apache.org/docs/format/Columnar.html#variable-size-list-layout
+/// [list layout]: https://arrow.apache.org/docs/format/Columnar.html#list-layout
+/// [list-view layout]: https://arrow.apache.org/docs/format/Columnar.html#listview-layout
 pub struct GenericListArray<OffsetSize: OffsetSizeTrait> {
     data_type: DataType,
     nulls: Option<NullBuffer>,
diff --git a/arrow-array/src/array/list_view_array.rs b/arrow-array/src/array/list_view_array.rs
index 7e52a6f3e..195ac7e11 100644
--- a/arrow-array/src/array/list_view_array.rs
+++ b/arrow-array/src/array/list_view_array.rs
@@ -32,16 +32,81 @@ pub type ListViewArray = GenericListViewArray<i32>;
 /// A [`GenericListViewArray`] of variable size lists, storing offsets as `i64`.
 pub type LargeListViewArray = GenericListViewArray<i64>;
 
+/// An array of [variable length lists], specifically in the [list-view layout].
 ///
-/// Different from [`crate::GenericListArray`] as it stores both an offset and length
-/// meaning that take / filter operations can be implemented without copying the underlying data.
+/// Differs from [`GenericListArray`] (which represents the [list layout]) in that
+/// the sizes of the child arrays are explicitly encoded in a separate buffer, instead
+/// of being derived from the difference between subsequent offsets in the offset buffer.
 ///
-/// [Variable-size List Layout: ListView Layout]: https://arrow.apache.org/docs/format/Columnar.html#listview-layout
+/// This allows the offsets (and subsequently child data) to be out of order. It also
+/// allows take / filter operations to be implemented without copying the underlying data.
+///
+/// # Representation
+///
+/// Given the same example array from [`GenericListArray`], it would be represented
+/// as such via a list-view layout array:
+///
+/// ```text
+///                                         ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+///                                                                         ┌ ─ ─ ─ ─ ─ ─ ┐    │
+///  ┌─────────────┐  ┌───────┐             │     ┌───┐   ┌───┐   ┌───┐       ┌───┐ ┌───┐
+///  │   [A,B,C]   │  │ (0,3) │                   │ 1 │   │ 0 │   │ 3 │     │ │ 1 │ │ A │ │ 0  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │      []     │  │ (3,0) │                   │ 1 │   │ 3 │   │ 0 │     │ │ 1 │ │ B │ │ 1  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │    NULL     │  │ (?,?) │                   │ 0 │   │ ? │   │ ? │     │ │ 1 │ │ C │ │ 2  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │     [D]     │  │ (4,1) │                   │ 1 │   │ 4 │   │ 1 │     │ │ ? │ │ ? │ │ 3  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │  [NULL, F]  │  │ (5,2) │                   │ 1 │   │ 5 │   │ 2 │     │ │ 1 │ │ D │ │ 4  │
+///  └─────────────┘  └───────┘             │     └───┘   └───┘   └───┘       ├───┤ ├───┤
+///                                                                         │ │ 0 │ │ ? │ │ 5  │
+///     Logical       Logical               │  Validity  Offsets  Sizes       ├───┤ ├───┤
+///      Values       Offset                   (nulls)                      │ │ 1 │ │ F │ │ 6  │
+///                   & Size                │                                 └───┘ └───┘
+///                                                                         │    Values   │    │
+///                 (offsets[i],            │   ListViewArray                   (Array)
+///                  sizes[i])                                              └ ─ ─ ─ ─ ─ ─ ┘    │
+///                                         └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+/// ```
+///
+/// Another way of representing the same array but taking advantage of the offsets being out of order:
+///
+/// ```text
+///                                         ┌ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+///                                                                         ┌ ─ ─ ─ ─ ─ ─ ┐    │
+///  ┌─────────────┐  ┌───────┐             │     ┌───┐   ┌───┐   ┌───┐       ┌───┐ ┌───┐
+///  │   [A,B,C]   │  │ (2,3) │                   │ 1 │   │ 2 │   │ 3 │     │ │ 0 │ │ ? │ │ 0  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │      []     │  │ (0,0) │                   │ 1 │   │ 0 │   │ 0 │     │ │ 1 │ │ F │ │ 1  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │    NULL     │  │ (?,?) │                   │ 0 │   │ ? │   │ ? │     │ │ 1 │ │ A │ │ 2  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │     [D]     │  │ (5,1) │                   │ 1 │   │ 5 │   │ 1 │     │ │ 1 │ │ B │ │ 3  │
+///  ├─────────────┤  ├───────┤             │     ├───┤   ├───┤   ├───┤       ├───┤ ├───┤
+///  │  [NULL, F]  │  │ (0,2) │                   │ 1 │   │ 0 │   │ 2 │     │ │ 1 │ │ C │ │ 4  │
+///  └─────────────┘  └───────┘             │     └───┘   └───┘   └───┘       ├───┤ ├───┤
+///                                                                         │ │ 1 │ │ D │ │ 5  │
+///     Logical       Logical               │  Validity  Offsets  Sizes       └───┘ └───┘
+///      Values       Offset                   (nulls)                      │    Values   │    │
+///                   & Size                │                                   (Array)  
+///                                                                         └ ─ ─ ─ ─ ─ ─ ┘    │
+///                 (offsets[i],            │   ListViewArray                          
+///                  sizes[i])                                                                 │
+///                                         └ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─
+/// ```
+///
+/// [`GenericListArray`]: crate::array::GenericListArray
+/// [variable length lists]: https://arrow.apache.org/docs/format/Columnar.html#variable-size-list-layout
+/// [list layout]: https://arrow.apache.org/docs/format/Columnar.html#list-layout
+/// [list-view layout]: https://arrow.apache.org/docs/format/Columnar.html#listview-layout
 #[derive(Clone)]
 pub struct GenericListViewArray<OffsetSize: OffsetSizeTrait> {
     data_type: DataType,
     nulls: Option<NullBuffer>,
     values: ArrayRef,
+    // Unlike GenericListArray, we do not use OffsetBuffer here as offsets are not
+    // guaranteed to be monotonically increasing.
     value_offsets: ScalarBuffer<OffsetSize>,
     value_sizes: ScalarBuffer<OffsetSize>,
 }
diff --git a/arrow-array/src/array/mod.rs b/arrow-array/src/array/mod.rs
index 23b3cb628..e41a3a1d7 100644
--- a/arrow-array/src/array/mod.rs
+++ b/arrow-array/src/array/mod.rs
@@ -620,6 +620,29 @@ impl<'a> StringArrayType<'a> for &'a StringViewArray {
     }
 }
 
+/// A trait for Arrow String Arrays, currently three types are supported:
+/// - `BinaryArray`
+/// - `LargeBinaryArray`
+/// - `BinaryViewArray`
+///
+/// This trait helps to abstract over the different types of binary arrays
+/// so that we don't need to duplicate the implementation for each type.
+pub trait BinaryArrayType<'a>: ArrayAccessor<Item = &'a [u8]> + Sized {
+    /// Constructs a new iterator
+    fn iter(&self) -> ArrayIter<Self>;
+}
+
+impl<'a, O: OffsetSizeTrait> BinaryArrayType<'a> for &'a GenericBinaryArray<O> {
+    fn iter(&self) -> ArrayIter<Self> {
+        GenericBinaryArray::<O>::iter(self)
+    }
+}
+impl<'a> BinaryArrayType<'a> for &'a BinaryViewArray {
+    fn iter(&self) -> ArrayIter<Self> {
+        BinaryViewArray::iter(self)
+    }
+}
+
 impl PartialEq for dyn Array + '_ {
     fn eq(&self, other: &Self) -> bool {
         self.to_data().eq(&other.to_data())
diff --git a/arrow-array/src/builder/mod.rs b/arrow-array/src/builder/mod.rs
index 29d75024e..e859f3794 100644
--- a/arrow-array/src/builder/mod.rs
+++ b/arrow-array/src/builder/mod.rs
@@ -216,8 +216,24 @@
 //!     RecordBatch::from(&builder.finish())
 //! }
 //! ```
+//!
+//! # Null / Validity Masks
+//!
+//! The [`NullBufferBuilder`] is optimized for creating the null mask for an array.
+//!
+//! ```
+//! # use arrow_array::builder::NullBufferBuilder;
+//! let mut builder = NullBufferBuilder::new(8);
+//! let mut builder = NullBufferBuilder::new(8);
+//! builder.append_n_non_nulls(7);
+//! builder.append_null();
+//! let buffer = builder.finish().unwrap();
+//! assert_eq!(buffer.len(), 8);
+//! assert_eq!(buffer.iter().collect::<Vec<_>>(), vec![true, true, true, true, true, true, true, false]);
+//! ```
 
 pub use arrow_buffer::BooleanBufferBuilder;
+pub use arrow_buffer::NullBufferBuilder;
 
 mod boolean_builder;
 pub use boolean_builder::*;
diff --git a/arrow-array/src/record_batch.rs b/arrow-array/src/record_batch.rs
index 8958ca6fa..a6c2aee7c 100644
--- a/arrow-array/src/record_batch.rs
+++ b/arrow-array/src/record_batch.rs
@@ -18,8 +18,9 @@
 //! A two-dimensional batch of column-oriented data with a defined
 //! [schema](arrow_schema::Schema).
 
+use crate::cast::AsArray;
 use crate::{new_empty_array, Array, ArrayRef, StructArray};
-use arrow_schema::{ArrowError, DataType, Field, Schema, SchemaBuilder, SchemaRef};
+use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema, SchemaBuilder, SchemaRef};
 use std::ops::Index;
 use std::sync::Arc;
 
@@ -394,6 +395,108 @@ impl RecordBatch {
         )
     }
 
+    /// Normalize a semi-structured [`RecordBatch`] into a flat table.
+    ///
+    /// Nested [`Field`]s will generate names separated by `separator`, up to a depth of `max_level`
+    /// (unlimited if `None`).
+    ///
+    /// e.g. given a [`RecordBatch`] with schema:
+    ///
+    /// ```text
+    ///     "foo": StructArray<"bar": Utf8>
+    /// ```
+    ///
+    /// A separator of `"."` would generate a batch with the schema:
+    ///
+    /// ```text
+    ///     "foo.bar": Utf8
+    /// ```
+    ///
+    /// Note that giving a depth of `Some(0)` to `max_level` is the same as passing in `None`;
+    /// it will be treated as unlimited.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use std::sync::Arc;
+    /// # use arrow_array::{ArrayRef, Int64Array, StringArray, StructArray, RecordBatch};
+    /// # use arrow_schema::{DataType, Field, Fields, Schema};
+    /// #
+    /// let animals: ArrayRef = Arc::new(StringArray::from(vec!["Parrot", ""]));
+    /// let n_legs: ArrayRef = Arc::new(Int64Array::from(vec![Some(2), Some(4)]));
+    ///
+    /// let animals_field = Arc::new(Field::new("animals", DataType::Utf8, true));
+    /// let n_legs_field = Arc::new(Field::new("n_legs", DataType::Int64, true));
+    ///
+    /// let a = Arc::new(StructArray::from(vec![
+    ///     (animals_field.clone(), Arc::new(animals.clone()) as ArrayRef),
+    ///     (n_legs_field.clone(), Arc::new(n_legs.clone()) as ArrayRef),
+    /// ]));
+    ///
+    /// let schema = Schema::new(vec![
+    ///     Field::new(
+    ///         "a",
+    ///         DataType::Struct(Fields::from(vec![animals_field, n_legs_field])),
+    ///         false,
+    ///     )
+    /// ]);
+    ///
+    /// let normalized = RecordBatch::try_new(Arc::new(schema), vec![a])
+    ///     .expect("valid conversion")
+    ///     .normalize(".", None)
+    ///     .expect("valid normalization");
+    ///
+    /// let expected = RecordBatch::try_from_iter_with_nullable(vec![
+    ///     ("a.animals", animals.clone(), true),
+    ///     ("a.n_legs", n_legs.clone(), true),
+    /// ])
+    /// .expect("valid conversion");
+    ///
+    /// assert_eq!(expected, normalized);
+    /// ```
+    pub fn normalize(&self, separator: &str, max_level: Option<usize>) -> Result<Self, ArrowError> {
+        let max_level = match max_level.unwrap_or(usize::MAX) {
+            0 => usize::MAX,
+            val => val,
+        };
+        let mut stack: Vec<(usize, &ArrayRef, Vec<&str>, &FieldRef)> = self
+            .columns
+            .iter()
+            .zip(self.schema.fields())
+            .rev()
+            .map(|(c, f)| {
+                let name_vec: Vec<&str> = vec![f.name()];
+                (0, c, name_vec, f)
+            })
+            .collect();
+        let mut columns: Vec<ArrayRef> = Vec::new();
+        let mut fields: Vec<FieldRef> = Vec::new();
+
+        while let Some((depth, c, name, field_ref)) = stack.pop() {
+            match field_ref.data_type() {
+                DataType::Struct(ff) if depth < max_level => {
+                    // Need to zip these in reverse to maintain original order
+                    for (cff, fff) in c.as_struct().columns().iter().zip(ff.into_iter()).rev() {
+                        let mut name = name.clone();
+                        name.push(separator);
+                        name.push(fff.name());
+                        stack.push((depth + 1, cff, name, fff))
+                    }
+                }
+                _ => {
+                    let updated_field = Field::new(
+                        name.concat(),
+                        field_ref.data_type().clone(),
+                        field_ref.is_nullable(),
+                    );
+                    columns.push(c.clone());
+                    fields.push(Arc::new(updated_field));
+                }
+            }
+        }
+        RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)
+    }
+
     /// Returns the number of columns in the record batch.
     ///
     /// # Example
@@ -768,8 +871,6 @@ where
 
 #[cfg(test)]
 mod tests {
-    use std::collections::HashMap;
-
     use super::*;
     use crate::{
         BooleanArray, Int32Array, Int64Array, Int8Array, ListArray, StringArray, StringViewArray,
@@ -777,6 +878,7 @@ mod tests {
     use arrow_buffer::{Buffer, ToByteSlice};
     use arrow_data::{ArrayData, ArrayDataBuilder};
     use arrow_schema::Fields;
+    use std::collections::HashMap;
 
     #[test]
     fn create_record_batch() {
@@ -1197,6 +1299,181 @@ mod tests {
         assert_ne!(batch1, batch2);
     }
 
+    #[test]
+    fn normalize_simple() {
+        let animals: ArrayRef = Arc::new(StringArray::from(vec!["Parrot", ""]));
+        let n_legs: ArrayRef = Arc::new(Int64Array::from(vec![Some(2), Some(4)]));
+        let year: ArrayRef = Arc::new(Int64Array::from(vec![None, Some(2022)]));
+
+        let animals_field = Arc::new(Field::new("animals", DataType::Utf8, true));
+        let n_legs_field = Arc::new(Field::new("n_legs", DataType::Int64, true));
+        let year_field = Arc::new(Field::new("year", DataType::Int64, true));
+
+        let a = Arc::new(StructArray::from(vec![
+            (animals_field.clone(), Arc::new(animals.clone()) as ArrayRef),
+            (n_legs_field.clone(), Arc::new(n_legs.clone()) as ArrayRef),
+            (year_field.clone(), Arc::new(year.clone()) as ArrayRef),
+        ]));
+
+        let month = Arc::new(Int64Array::from(vec![Some(4), Some(6)]));
+
+        let schema = Schema::new(vec![
+            Field::new(
+                "a",
+                DataType::Struct(Fields::from(vec![animals_field, n_legs_field, year_field])),
+                false,
+            ),
+            Field::new("month", DataType::Int64, true),
+        ]);
+
+        let normalized =
+            RecordBatch::try_new(Arc::new(schema.clone()), vec![a.clone(), month.clone()])
+                .expect("valid conversion")
+                .normalize(".", Some(0))
+                .expect("valid normalization");
+
+        let expected = RecordBatch::try_from_iter_with_nullable(vec![
+            ("a.animals", animals.clone(), true),
+            ("a.n_legs", n_legs.clone(), true),
+            ("a.year", year.clone(), true),
+            ("month", month.clone(), true),
+        ])
+        .expect("valid conversion");
+
+        assert_eq!(expected, normalized);
+
+        // check 0 and None have the same effect
+        let normalized = RecordBatch::try_new(Arc::new(schema), vec![a, month.clone()])
+            .expect("valid conversion")
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        assert_eq!(expected, normalized);
+    }
+
+    #[test]
+    fn normalize_nested() {
+        // Initialize schema
+        let a = Arc::new(Field::new("a", DataType::Int64, true));
+        let b = Arc::new(Field::new("b", DataType::Int64, false));
+        let c = Arc::new(Field::new("c", DataType::Int64, true));
+
+        let one = Arc::new(Field::new(
+            "1",
+            DataType::Struct(Fields::from(vec![a.clone(), b.clone(), c.clone()])),
+            false,
+        ));
+        let two = Arc::new(Field::new(
+            "2",
+            DataType::Struct(Fields::from(vec![a.clone(), b.clone(), c.clone()])),
+            true,
+        ));
+
+        let exclamation = Arc::new(Field::new(
+            "!",
+            DataType::Struct(Fields::from(vec![one.clone(), two.clone()])),
+            false,
+        ));
+
+        let schema = Schema::new(vec![exclamation.clone()]);
+
+        // Initialize fields
+        let a_field = Int64Array::from(vec![Some(0), Some(1)]);
+        let b_field = Int64Array::from(vec![Some(2), Some(3)]);
+        let c_field = Int64Array::from(vec![None, Some(4)]);
+
+        let one_field = StructArray::from(vec![
+            (a.clone(), Arc::new(a_field.clone()) as ArrayRef),
+            (b.clone(), Arc::new(b_field.clone()) as ArrayRef),
+            (c.clone(), Arc::new(c_field.clone()) as ArrayRef),
+        ]);
+        let two_field = StructArray::from(vec![
+            (a.clone(), Arc::new(a_field.clone()) as ArrayRef),
+            (b.clone(), Arc::new(b_field.clone()) as ArrayRef),
+            (c.clone(), Arc::new(c_field.clone()) as ArrayRef),
+        ]);
+
+        let exclamation_field = Arc::new(StructArray::from(vec![
+            (one.clone(), Arc::new(one_field) as ArrayRef),
+            (two.clone(), Arc::new(two_field) as ArrayRef),
+        ]));
+
+        // Normalize top level
+        let normalized =
+            RecordBatch::try_new(Arc::new(schema.clone()), vec![exclamation_field.clone()])
+                .expect("valid conversion")
+                .normalize(".", Some(1))
+                .expect("valid normalization");
+
+        let expected = RecordBatch::try_from_iter_with_nullable(vec![
+            (
+                "!.1",
+                Arc::new(StructArray::from(vec![
+                    (a.clone(), Arc::new(a_field.clone()) as ArrayRef),
+                    (b.clone(), Arc::new(b_field.clone()) as ArrayRef),
+                    (c.clone(), Arc::new(c_field.clone()) as ArrayRef),
+                ])) as ArrayRef,
+                false,
+            ),
+            (
+                "!.2",
+                Arc::new(StructArray::from(vec![
+                    (a.clone(), Arc::new(a_field.clone()) as ArrayRef),
+                    (b.clone(), Arc::new(b_field.clone()) as ArrayRef),
+                    (c.clone(), Arc::new(c_field.clone()) as ArrayRef),
+                ])) as ArrayRef,
+                true,
+            ),
+        ])
+        .expect("valid conversion");
+
+        assert_eq!(expected, normalized);
+
+        // Normalize all levels
+        let normalized = RecordBatch::try_new(Arc::new(schema), vec![exclamation_field])
+            .expect("valid conversion")
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        let expected = RecordBatch::try_from_iter_with_nullable(vec![
+            ("!.1.a", Arc::new(a_field.clone()) as ArrayRef, true),
+            ("!.1.b", Arc::new(b_field.clone()) as ArrayRef, false),
+            ("!.1.c", Arc::new(c_field.clone()) as ArrayRef, true),
+            ("!.2.a", Arc::new(a_field.clone()) as ArrayRef, true),
+            ("!.2.b", Arc::new(b_field.clone()) as ArrayRef, false),
+            ("!.2.c", Arc::new(c_field.clone()) as ArrayRef, true),
+        ])
+        .expect("valid conversion");
+
+        assert_eq!(expected, normalized);
+    }
+
+    #[test]
+    fn normalize_empty() {
+        let animals_field = Arc::new(Field::new("animals", DataType::Utf8, true));
+        let n_legs_field = Arc::new(Field::new("n_legs", DataType::Int64, true));
+        let year_field = Arc::new(Field::new("year", DataType::Int64, true));
+
+        let schema = Schema::new(vec![
+            Field::new(
+                "a",
+                DataType::Struct(Fields::from(vec![animals_field, n_legs_field, year_field])),
+                false,
+            ),
+            Field::new("month", DataType::Int64, true),
+        ]);
+
+        let normalized = RecordBatch::new_empty(Arc::new(schema.clone()))
+            .normalize(".", Some(0))
+            .expect("valid normalization");
+
+        let expected = RecordBatch::new_empty(Arc::new(
+            schema.normalize(".", Some(0)).expect("valid normalization"),
+        ));
+
+        assert_eq!(expected, normalized);
+    }
+
     #[test]
     fn project() {
         let a: ArrayRef = Arc::new(Int32Array::from(vec![Some(1), None, Some(3)]));
diff --git a/arrow-buffer/src/builder/boolean.rs b/arrow-buffer/src/builder/boolean.rs
index 67b306dd8..0424e128d 100644
--- a/arrow-buffer/src/builder/boolean.rs
+++ b/arrow-buffer/src/builder/boolean.rs
@@ -32,7 +32,11 @@ pub struct BooleanBufferBuilder {
 }
 
 impl BooleanBufferBuilder {
-    /// Creates a new `BooleanBufferBuilder`
+    /// Creates a new `BooleanBufferBuilder` with sufficient space for
+    /// `capacity` bits (not bytes).
+    ///
+    /// The capacity is rounded up to the nearest multiple of 8 for the
+    /// allocation.
     #[inline]
     pub fn new(capacity: usize) -> Self {
         let byte_capacity = bit_util::ceil(capacity, 8);
diff --git a/arrow-buffer/src/builder/null.rs b/arrow-buffer/src/builder/null.rs
index ddbca46f7..f88868db8 100644
--- a/arrow-buffer/src/builder/null.rs
+++ b/arrow-buffer/src/builder/null.rs
@@ -27,6 +27,23 @@ use crate::{BooleanBufferBuilder, MutableBuffer, NullBuffer};
 ///
 /// This optimization is **very** important for the performance as it avoids
 /// allocating memory for the null buffer when there are no nulls.
+///
+/// # Example
+/// ```
+/// # use arrow_buffer::NullBufferBuilder;
+/// let mut builder = NullBufferBuilder::new(8);
+/// builder.append_n_non_nulls(8);
+/// // If no non null values are appended, the null buffer is not created
+/// let buffer = builder.finish();
+/// assert!(buffer.is_none());
+/// // however, if a null value is appended, the null buffer is created
+/// let mut builder = NullBufferBuilder::new(8);
+/// builder.append_n_non_nulls(7);
+/// builder.append_null();
+/// let buffer = builder.finish().unwrap();
+/// assert_eq!(buffer.len(), 8);
+/// assert_eq!(buffer.iter().collect::<Vec<_>>(), vec![true, true, true, true, true, true, true, false]);
+/// ```
 #[derive(Debug)]
 pub struct NullBufferBuilder {
     bitmap_builder: Option<BooleanBufferBuilder>,
@@ -37,7 +54,10 @@ pub struct NullBufferBuilder {
 
 impl NullBufferBuilder {
     /// Creates a new empty builder.
-    /// `capacity` is the number of bits in the null buffer.
+    ///
+    /// Note that this method does not allocate any memory, regardless of the
+    /// `capacity` parameter. If an allocation is required, `capacity` is the
+    /// size in bits (not bytes) that will be allocated at minimum.
     pub fn new(capacity: usize) -> Self {
         Self {
             bitmap_builder: None,
diff --git a/arrow-data/src/decimal.rs b/arrow-data/src/decimal.rs
index fe19db641..5453fb120 100644
--- a/arrow-data/src/decimal.rs
+++ b/arrow-data/src/decimal.rs
@@ -15,10 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Defines maximum and minimum values for `decimal256` and `decimal128` types for varying precisions.
+//! Maximum and minimum values for [`Decimal256`] and [`Decimal128`].
 //!
-//! Also provides functions to validate if a given decimal value is within the valid range of the decimal type.
-
+//! Also provides functions to validate if a given decimal value is within
+//! the valid range of the decimal type.
+//!
+//! [`Decimal128`]: arrow_schema::DataType::Decimal128
+//! [`Decimal256`]: arrow_schema::DataType::Decimal256
 use arrow_buffer::i256;
 use arrow_schema::ArrowError;
 
@@ -27,12 +30,27 @@ pub use arrow_schema::{
     DECIMAL_DEFAULT_SCALE,
 };
 
-/// MAX decimal256 value of little-endian format for each precision.
-/// Each element is the max value of signed 256-bit integer for the specified precision which
-/// is encoded to the 32-byte width format of little-endian.
+/// `MAX_DECIMAL256_FOR_EACH_PRECISION[p]` holds the maximum [`i256`] value that can
+/// be stored in a [`Decimal256`] value of precision `p`.
+///
+/// # Notes
+///
+/// Each element is the max value of signed 256-bit integer for the specified
+/// precision which is encoded to the 32-byte width format of little-endian.
+///
 /// The first element is unused and is inserted so that we can look up using
 /// precision as the index without the need to subtract 1 first.
-pub(crate) const MAX_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION: [i256; 77] = [
+///
+/// # Example
+/// ```
+/// # use arrow_buffer::i256;
+/// # use arrow_data::decimal::MAX_DECIMAL256_FOR_EACH_PRECISION;
+/// assert_eq!(MAX_DECIMAL256_FOR_EACH_PRECISION[3], i256::from(999));
+/// ```
+///
+/// [`Decimal256`]: arrow_schema::DataType::Decimal256
+/// [`i256`]: arrow_buffer::i256
+pub const MAX_DECIMAL256_FOR_EACH_PRECISION: [i256; 77] = [
     i256::from_i128(0_i128), // unused first element
     i256::from_le_bytes([
         9, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
@@ -340,12 +358,26 @@ pub(crate) const MAX_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION: [i256; 77] = [
     ]),
 ];
 
-/// MIN decimal256 value of little-endian format for each precision.
+/// `MIN_DECIMAL256_FOR_EACH_PRECISION[p]` holds the minimum [`i256`] value that can
+/// be stored in a [`Decimal256`] value of precision `p`.
+///
+/// # Notes
+///
 /// Each element is the min value of signed 256-bit integer for the specified precision which
 /// is encoded to the 76-byte width format of little-endian.
+///
 /// The first element is unused and is inserted so that we can look up using
 /// precision as the index without the need to subtract 1 first.
-pub(crate) const MIN_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION: [i256; 77] = [
+/// # Example
+/// ```
+/// # use arrow_buffer::i256;
+/// # use arrow_data::decimal::MIN_DECIMAL256_FOR_EACH_PRECISION;
+/// assert_eq!(MIN_DECIMAL256_FOR_EACH_PRECISION[3], i256::from(-999));
+/// ```
+///
+/// [`i256`]: arrow_buffer::i256
+/// [`Decimal256`]: arrow_schema::DataType::Decimal256
+pub const MIN_DECIMAL256_FOR_EACH_PRECISION: [i256; 77] = [
     i256::from_i128(0_i128), // unused first element
     i256::from_le_bytes([
         247, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255, 255,
@@ -654,7 +686,13 @@ pub(crate) const MIN_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION: [i256; 77] = [
 ];
 
 /// `MAX_DECIMAL_FOR_EACH_PRECISION[p-1]` holds the maximum `i128` value that can
-/// be stored in [arrow_schema::DataType::Decimal128] value of precision `p`
+/// be stored in a [`Decimal128`] value of precision `p`
+///
+/// [`Decimal128`]: arrow_schema::DataType::Decimal128
+#[deprecated(
+    since = "54.1.0",
+    note = "Use MAX_DECIMAL128_FOR_EACH_PRECISION (note indexes are different)"
+)]
 #[allow(dead_code)] // no longer used but is part of our public API
 pub const MAX_DECIMAL_FOR_EACH_PRECISION: [i128; 38] = [
     9,
@@ -698,8 +736,14 @@ pub const MAX_DECIMAL_FOR_EACH_PRECISION: [i128; 38] = [
 ];
 
 /// `MIN_DECIMAL_FOR_EACH_PRECISION[p-1]` holds the minimum `i128` value that can
-/// be stored in a [arrow_schema::DataType::Decimal128] value of precision `p`
+/// be stored in a [`Decimal128`] value of precision `p`
+///
+/// [`Decimal128`]: arrow_schema::DataType::Decimal128
 #[allow(dead_code)] // no longer used but is part of our public API
+#[deprecated(
+    since = "54.1.0",
+    note = "Use MIN_DECIMAL_FOR_EACH_PRECISION (note indexes are different)"
+)]
 pub const MIN_DECIMAL_FOR_EACH_PRECISION: [i128; 38] = [
     -9,
     -99,
@@ -741,11 +785,22 @@ pub const MIN_DECIMAL_FOR_EACH_PRECISION: [i128; 38] = [
     -99999999999999999999999999999999999999,
 ];
 
-/// `MAX_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[p]` holds the maximum `i128` value that can
-/// be stored in [arrow_schema::DataType::Decimal128] value of precision `p`.
+/// `MAX_DECIMAL128_FOR_EACH_PRECISION[p]` holds the maximum `i128` value that can
+/// be stored in [`Decimal128`] value of precision `p`.
+///
+/// # Notes
+///
 /// The first element is unused and is inserted so that we can look up using
 /// precision as the index without the need to subtract 1 first.
-pub(crate) const MAX_DECIMAL_FOR_EACH_PRECISION_ONE_BASED: [i128; 39] = [
+///
+/// # Example
+/// ```
+/// # use arrow_data::decimal::MAX_DECIMAL128_FOR_EACH_PRECISION;
+/// assert_eq!(MAX_DECIMAL128_FOR_EACH_PRECISION[3], 999);
+/// ```
+///
+/// [`Decimal128`]: arrow_schema::DataType::Decimal128
+pub const MAX_DECIMAL128_FOR_EACH_PRECISION: [i128; 39] = [
     0, // unused first element
     9,
     99,
@@ -788,10 +843,21 @@ pub(crate) const MAX_DECIMAL_FOR_EACH_PRECISION_ONE_BASED: [i128; 39] = [
 ];
 
 /// `MIN_DECIMAL_FOR_EACH_PRECISION[p]` holds the minimum `i128` value that can
-/// be stored in a [arrow_schema::DataType::Decimal128] value of precision `p`.
+/// be stored in a [`Decimal128`] value of precision `p`.
+///
+/// # Notes
+///
 /// The first element is unused and is inserted so that we can look up using
 /// precision as the index without the need to subtract 1 first.
-pub(crate) const MIN_DECIMAL_FOR_EACH_PRECISION_ONE_BASED: [i128; 39] = [
+///
+/// # Example
+/// ```
+/// # use arrow_data::decimal::MIN_DECIMAL128_FOR_EACH_PRECISION;
+/// assert_eq!(MIN_DECIMAL128_FOR_EACH_PRECISION[3], -999);
+/// ```
+///
+/// [`Decimal128`]: arrow_schema::DataType::Decimal128
+pub const MIN_DECIMAL128_FOR_EACH_PRECISION: [i128; 39] = [
     0, // unused first element
     -9,
     -99,
@@ -834,7 +900,9 @@ pub(crate) const MIN_DECIMAL_FOR_EACH_PRECISION_ONE_BASED: [i128; 39] = [
 ];
 
 /// Validates that the specified `i128` value can be properly
-/// interpreted as a Decimal number with precision `precision`
+/// interpreted as a [`Decimal128`] number with precision `precision`
+///
+/// [`Decimal128`]: arrow_schema::DataType::Decimal128
 #[inline]
 pub fn validate_decimal_precision(value: i128, precision: u8) -> Result<(), ArrowError> {
     if precision > DECIMAL128_MAX_PRECISION {
@@ -842,32 +910,36 @@ pub fn validate_decimal_precision(value: i128, precision: u8) -> Result<(), Arro
             "Max precision of a Decimal128 is {DECIMAL128_MAX_PRECISION}, but got {precision}",
         )));
     }
-    if value > MAX_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[precision as usize] {
+    if value > MAX_DECIMAL128_FOR_EACH_PRECISION[precision as usize] {
         Err(ArrowError::InvalidArgumentError(format!(
             "{value} is too large to store in a Decimal128 of precision {precision}. Max is {}",
-            MAX_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[precision as usize]
+            MAX_DECIMAL128_FOR_EACH_PRECISION[precision as usize]
         )))
-    } else if value < MIN_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[precision as usize] {
+    } else if value < MIN_DECIMAL128_FOR_EACH_PRECISION[precision as usize] {
         Err(ArrowError::InvalidArgumentError(format!(
             "{value} is too small to store in a Decimal128 of precision {precision}. Min is {}",
-            MIN_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[precision as usize]
+            MIN_DECIMAL128_FOR_EACH_PRECISION[precision as usize]
         )))
     } else {
         Ok(())
     }
 }
 
-/// Determines whether the specified `i128` value can be properly
-/// interpreted as a Decimal number with precision `precision`
+/// Returns true if the specified `i128` value can be properly
+/// interpreted as a [`Decimal128`] number with precision `precision`
+///
+/// [`Decimal128`]: arrow_schema::DataType::Decimal128
 #[inline]
 pub fn is_validate_decimal_precision(value: i128, precision: u8) -> bool {
     precision <= DECIMAL128_MAX_PRECISION
-        && value >= MIN_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[precision as usize]
-        && value <= MAX_DECIMAL_FOR_EACH_PRECISION_ONE_BASED[precision as usize]
+        && value >= MIN_DECIMAL128_FOR_EACH_PRECISION[precision as usize]
+        && value <= MAX_DECIMAL128_FOR_EACH_PRECISION[precision as usize]
 }
 
 /// Validates that the specified `i256` of value can be properly
-/// interpreted as a Decimal256 number with precision `precision`
+/// interpreted as a [`Decimal256`] number with precision `precision`
+///
+/// [`Decimal256`]: arrow_schema::DataType::Decimal256
 #[inline]
 pub fn validate_decimal256_precision(value: i256, precision: u8) -> Result<(), ArrowError> {
     if precision > DECIMAL256_MAX_PRECISION {
@@ -875,26 +947,28 @@ pub fn validate_decimal256_precision(value: i256, precision: u8) -> Result<(), A
             "Max precision of a Decimal256 is {DECIMAL256_MAX_PRECISION}, but got {precision}",
         )));
     }
-    if value > MAX_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION[precision as usize] {
+    if value > MAX_DECIMAL256_FOR_EACH_PRECISION[precision as usize] {
         Err(ArrowError::InvalidArgumentError(format!(
             "{value:?} is too large to store in a Decimal256 of precision {precision}. Max is {:?}",
-            MAX_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION[precision as usize]
+            MAX_DECIMAL256_FOR_EACH_PRECISION[precision as usize]
         )))
-    } else if value < MIN_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION[precision as usize] {
+    } else if value < MIN_DECIMAL256_FOR_EACH_PRECISION[precision as usize] {
         Err(ArrowError::InvalidArgumentError(format!(
             "{value:?} is too small to store in a Decimal256 of precision {precision}. Min is {:?}",
-            MIN_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION[precision as usize]
+            MIN_DECIMAL256_FOR_EACH_PRECISION[precision as usize]
         )))
     } else {
         Ok(())
     }
 }
 
-/// Determines whether the specified `i256` value can be properly
-/// interpreted as a Decimal256 number with precision `precision`
+/// Return true if the specified `i256` value can be properly
+/// interpreted as a [`Decimal256`] number with precision `precision`
+///
+/// [`Decimal256`]: arrow_schema::DataType::Decimal256
 #[inline]
 pub fn is_validate_decimal256_precision(value: i256, precision: u8) -> bool {
     precision <= DECIMAL256_MAX_PRECISION
-        && value >= MIN_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION[precision as usize]
-        && value <= MAX_DECIMAL_BYTES_FOR_LARGER_EACH_PRECISION[precision as usize]
+        && value >= MIN_DECIMAL256_FOR_EACH_PRECISION[precision as usize]
+        && value <= MAX_DECIMAL256_FOR_EACH_PRECISION[precision as usize]
 }
diff --git a/arrow-ipc/src/lib.rs b/arrow-ipc/src/lib.rs
index a76083b93..4638abdb4 100644
--- a/arrow-ipc/src/lib.rs
+++ b/arrow-ipc/src/lib.rs
@@ -17,7 +17,26 @@
 
 //! Support for the [Arrow IPC Format]
 //!
+//! The Arrow IPC format defines how to read and write [`RecordBatch`]es to/from
+//! a file or stream of bytes. This format can be used to serialize and deserialize
+//! data to files and over the network.
+//!
+//! There are two variants of the IPC format:
+//! 1. [IPC Streaming Format]: Supports streaming data sources, implemented by
+//!    [StreamReader] and [StreamWriter]
+//!
+//! 2. [IPC File Format]: Supports random access, implemented by [FileReader] and
+//!    [FileWriter].
+//!
+//! See the [`reader`] and [`writer`] modules for more information.
+//!
 //! [Arrow IPC Format]: https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc
+//! [IPC Streaming Format]: https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
+//! [StreamReader]: reader::StreamReader
+//! [StreamWriter]: writer::StreamWriter
+//! [IPC File Format]: https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format
+//! [FileReader]: reader::FileReader
+//! [FileWriter]: writer::FileWriter
 
 #![warn(missing_docs)]
 pub mod convert;
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index 5cf208c8e..b72785651 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -17,8 +17,12 @@
 
 //! Arrow IPC File and Stream Readers
 //!
-//! The `FileReader` and `StreamReader` have similar interfaces,
-//! however the `FileReader` expects a reader that supports `Seek`ing
+//! # Notes
+//!
+//! The [`FileReader`] and [`StreamReader`] have similar interfaces,
+//! however the [`FileReader`] expects a reader that supports [`Seek`]ing
+//!
+//! [`Seek`]: std::io::Seek
 
 mod stream;
 
@@ -683,6 +687,10 @@ pub fn read_footer_length(buf: [u8; 10]) -> Result<usize, ArrowError> {
 ///
 /// For a higher-level interface see [`FileReader`]
 ///
+/// For an example of using this API with `mmap` see the [`zero_copy_ipc`] example.
+///
+/// [`zero_copy_ipc`]: https://github.com/apache/arrow-rs/blob/main/arrow/examples/zero_copy_ipc.rs
+///
 /// ```
 /// # use std::sync::Arc;
 /// # use arrow_array::*;
@@ -993,7 +1001,50 @@ impl FileReaderBuilder {
     }
 }
 
-/// Arrow File reader
+/// Arrow File Reader
+///
+/// Reads Arrow [`RecordBatch`]es from bytes in the [IPC File Format],
+/// providing random access to the record batches.
+///
+/// # See Also
+///
+/// * [`Self::set_index`] for random access
+/// * [`StreamReader`] for reading streaming data
+///
+/// # Example: Reading from a `File`
+/// ```
+/// # use std::io::Cursor;
+/// use arrow_array::record_batch;
+/// # use arrow_ipc::reader::FileReader;
+/// # use arrow_ipc::writer::FileWriter;
+/// # let batch = record_batch!(("a", Int32, [1, 2, 3])).unwrap();
+/// # let mut file = vec![]; // mimic a stream for the example
+/// # {
+/// #  let mut writer = FileWriter::try_new(&mut file, &batch.schema()).unwrap();
+/// #  writer.write(&batch).unwrap();
+/// #  writer.write(&batch).unwrap();
+/// #  writer.finish().unwrap();
+/// # }
+/// # let mut file = Cursor::new(&file);
+/// let projection = None; // read all columns
+/// let mut reader = FileReader::try_new(&mut file, projection).unwrap();
+/// // Position the reader to the second batch
+/// reader.set_index(1).unwrap();
+/// // read batches from the reader using the Iterator trait
+/// let mut num_rows = 0;
+/// for batch in reader {
+///    let batch = batch.unwrap();
+///    num_rows += batch.num_rows();
+/// }
+/// assert_eq!(num_rows, 3);
+/// ```
+/// # Example: Reading from `mmap`ed file
+///
+/// For an example creating Arrays without copying using  memory mapped (`mmap`)
+/// files see the [`zero_copy_ipc`] example.
+///
+/// [IPC Streaming Format]: https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
+/// [`zero_copy_ipc`]: https://github.com/apache/arrow-rs/blob/main/arrow/examples/zero_copy_ipc.rs
 pub struct FileReader<R> {
     /// File reader that supports reading and seeking
     reader: R,
@@ -1070,7 +1121,7 @@ impl<R: Read + Seek> FileReader<R> {
         self.decoder.schema.clone()
     }
 
-    /// Read a specific record batch
+    /// See to a specific [`RecordBatch`]
     ///
     /// Sets the current block to the index, allowing random reads
     pub fn set_index(&mut self, index: usize) -> Result<(), ArrowError> {
@@ -1128,7 +1179,39 @@ impl<R: Read + Seek> RecordBatchReader for FileReader<R> {
     }
 }
 
-/// Arrow Stream reader
+/// Arrow Stream Reader
+///
+/// Reads Arrow [`RecordBatch`]es from bytes in the [IPC Streaming Format].
+///
+/// # See Also
+///
+/// * [`FileReader`] for random access.
+///
+/// # Example
+/// ```
+/// # use arrow_array::record_batch;
+/// # use arrow_ipc::reader::StreamReader;
+/// # use arrow_ipc::writer::StreamWriter;
+/// # let batch = record_batch!(("a", Int32, [1, 2, 3])).unwrap();
+/// # let mut stream = vec![]; // mimic a stream for the example
+/// # {
+/// #  let mut writer = StreamWriter::try_new(&mut stream, &batch.schema()).unwrap();
+/// #  writer.write(&batch).unwrap();
+/// #  writer.finish().unwrap();
+/// # }
+/// # let stream = stream.as_slice();
+/// let projection = None; // read all columns
+/// let mut reader = StreamReader::try_new(stream, projection).unwrap();
+/// // read batches from the reader using the Iterator trait
+/// let mut num_rows = 0;
+/// for batch in reader {
+///    let batch = batch.unwrap();
+///    num_rows += batch.num_rows();
+/// }
+/// assert_eq!(num_rows, 3);
+/// ```
+///
+/// [IPC Streaming Format]: https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
 pub struct StreamReader<R> {
     /// Stream reader
     reader: R,
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index ee5b9a54c..451b7e02c 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -17,8 +17,12 @@
 
 //! Arrow IPC File and Stream Writers
 //!
-//! The `FileWriter` and `StreamWriter` have similar interfaces,
-//! however the `FileWriter` expects a reader that supports `Seek`ing
+//! # Notes
+//!
+//! [`FileWriter`] and [`StreamWriter`] have similar interfaces,
+//! however the [`FileWriter`] expects a reader that supports [`Seek`]ing
+//!
+//! [`Seek`]: std::io::Seek
 
 use std::cmp::min;
 use std::collections::HashMap;
@@ -188,7 +192,7 @@ impl Default for IpcWriteOptions {
 /// Handles low level details of encoding [`Array`] and [`Schema`] into the
 /// [Arrow IPC Format].
 ///
-/// # Example:
+/// # Example
 /// ```
 /// # fn run() {
 /// # use std::sync::Arc;
@@ -905,7 +909,28 @@ impl DictionaryTracker {
     }
 }
 
-/// Writer for an IPC file
+/// Arrow File Writer
+///
+/// Writes Arrow [`RecordBatch`]es in the [IPC File Format].
+///
+/// # See Also
+///
+/// * [`StreamWriter`] for writing IPC Streams
+///
+/// # Example
+/// ```
+/// # use arrow_array::record_batch;
+/// # use arrow_ipc::writer::StreamWriter;
+/// # let mut file = vec![]; // mimic a file for the example
+/// let batch = record_batch!(("a", Int32, [1, 2, 3])).unwrap();
+/// // create a new writer, the schema must be known in advance
+/// let mut writer = StreamWriter::try_new(&mut file, &batch.schema()).unwrap();
+/// // write each batch to the underlying writer
+/// writer.write(&batch).unwrap();
+/// // When all batches are written, call finish to flush all buffers
+/// writer.finish().unwrap();
+/// ```
+/// [IPC File Format]: https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format
 pub struct FileWriter<W> {
     /// The object to write to
     writer: W,
@@ -1108,7 +1133,7 @@ impl<W: Write> FileWriter<W> {
         Ok(())
     }
 
-    /// Unwraps the the underlying writer.
+    /// Unwraps the underlying writer.
     ///
     /// The writer is flushed and the FileWriter is finished before returning.
     ///
@@ -1135,7 +1160,29 @@ impl<W: Write> RecordBatchWriter for FileWriter<W> {
     }
 }
 
-/// Writer for an IPC stream
+/// Arrow Stream Writer
+///
+/// Writes Arrow [`RecordBatch`]es to bytes using the [IPC Streaming Format].
+///
+/// # See Also
+///
+/// * [`FileWriter`] for writing IPC Files
+///
+/// # Example
+/// ```
+/// # use arrow_array::record_batch;
+/// # use arrow_ipc::writer::StreamWriter;
+/// # let mut stream = vec![]; // mimic a stream for the example
+/// let batch = record_batch!(("a", Int32, [1, 2, 3])).unwrap();
+/// // create a new writer, the schema must be known in advance
+/// let mut writer = StreamWriter::try_new(&mut stream, &batch.schema()).unwrap();
+/// // write each batch to the underlying stream
+/// writer.write(&batch).unwrap();
+/// // When all batches are written, call finish to flush all buffers
+/// writer.finish().unwrap();
+/// ```
+///
+/// [IPC Streaming Format]: https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format
 pub struct StreamWriter<W> {
     /// The object to write to
     writer: W,
@@ -1473,10 +1520,7 @@ fn reencode_offsets<O: OffsetSizeTrait>(
     let offsets = match start_offset.as_usize() {
         0 => {
             let size = size_of::<O>();
-            offsets.slice_with_length(
-                data.offset() * size,
-                (data.offset() + data.len() + 1) * size,
-            )
+            offsets.slice_with_length(data.offset() * size, (data.len() + 1) * size)
         }
         _ => offset_slice.iter().map(|x| *x - *start_offset).collect(),
     };
@@ -1793,9 +1837,9 @@ mod tests {
     use std::io::Cursor;
     use std::io::Seek;
 
-    use arrow_array::builder::GenericListBuilder;
     use arrow_array::builder::MapBuilder;
     use arrow_array::builder::UnionBuilder;
+    use arrow_array::builder::{GenericListBuilder, ListBuilder, StringBuilder};
     use arrow_array::builder::{PrimitiveRunBuilder, UInt32Builder};
     use arrow_array::types::*;
     use arrow_buffer::ScalarBuffer;
@@ -2433,6 +2477,126 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_large_slice_uint32() {
+        ensure_roundtrip(Arc::new(UInt32Array::from_iter((0..8000).map(|i| {
+            if i % 2 == 0 {
+                Some(i)
+            } else {
+                None
+            }
+        }))));
+    }
+
+    #[test]
+    fn test_large_slice_string() {
+        let strings: Vec<_> = (0..8000)
+            .map(|i| {
+                if i % 2 == 0 {
+                    Some(format!("value{}", i))
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        ensure_roundtrip(Arc::new(StringArray::from(strings)));
+    }
+
+    #[test]
+    fn test_large_slice_string_list() {
+        let mut ls = ListBuilder::new(StringBuilder::new());
+
+        let mut s = String::new();
+        for row_number in 0..8000 {
+            if row_number % 2 == 0 {
+                for list_element in 0..1000 {
+                    s.clear();
+                    use std::fmt::Write;
+                    write!(&mut s, "value{row_number}-{list_element}").unwrap();
+                    ls.values().append_value(&s);
+                }
+                ls.append(true)
+            } else {
+                ls.append(false); // null
+            }
+        }
+
+        ensure_roundtrip(Arc::new(ls.finish()));
+    }
+
+    #[test]
+    fn test_large_slice_string_list_of_lists() {
+        // The reason for the special test is to verify reencode_offsets which looks both at
+        // the starting offset and the data offset.  So need a dataset where the starting_offset
+        // is zero but the data offset is not.
+        let mut ls = ListBuilder::new(ListBuilder::new(StringBuilder::new()));
+
+        for _ in 0..4000 {
+            ls.values().append(true);
+            ls.append(true)
+        }
+
+        let mut s = String::new();
+        for row_number in 0..4000 {
+            if row_number % 2 == 0 {
+                for list_element in 0..1000 {
+                    s.clear();
+                    use std::fmt::Write;
+                    write!(&mut s, "value{row_number}-{list_element}").unwrap();
+                    ls.values().values().append_value(&s);
+                }
+                ls.values().append(true);
+                ls.append(true)
+            } else {
+                ls.append(false); // null
+            }
+        }
+
+        ensure_roundtrip(Arc::new(ls.finish()));
+    }
+
+    /// Read/write a record batch to a File and Stream and ensure it is the same at the outout
+    fn ensure_roundtrip(array: ArrayRef) {
+        let num_rows = array.len();
+        let orig_batch = RecordBatch::try_from_iter(vec![("a", array)]).unwrap();
+        // take off the first element
+        let sliced_batch = orig_batch.slice(1, num_rows - 1);
+
+        let schema = orig_batch.schema();
+        let stream_data = {
+            let mut writer = StreamWriter::try_new(vec![], &schema).unwrap();
+            writer.write(&sliced_batch).unwrap();
+            writer.into_inner().unwrap()
+        };
+        let read_batch = {
+            let projection = None;
+            let mut reader = StreamReader::try_new(Cursor::new(stream_data), projection).unwrap();
+            reader
+                .next()
+                .expect("expect no errors reading batch")
+                .expect("expect batch")
+        };
+        assert_eq!(sliced_batch, read_batch);
+
+        let file_data = {
+            let mut writer = FileWriter::try_new_buffered(vec![], &schema).unwrap();
+            writer.write(&sliced_batch).unwrap();
+            writer.into_inner().unwrap().into_inner().unwrap()
+        };
+        let read_batch = {
+            let projection = None;
+            let mut reader = FileReader::try_new(Cursor::new(file_data), projection).unwrap();
+            reader
+                .next()
+                .expect("expect no errors reading batch")
+                .expect("expect batch")
+        };
+        assert_eq!(sliced_batch, read_batch);
+
+        // TODO test file writer/reader
+    }
+
     #[test]
     fn encode_bools_slice() {
         // Test case for https://github.com/apache/arrow-rs/issues/3496
@@ -2615,6 +2779,40 @@ mod tests {
         builder.finish()
     }
 
+    #[test]
+    fn reencode_offsets_when_first_offset_is_not_zero() {
+        let original_list = generate_list_data::<i32>();
+        let original_data = original_list.into_data();
+        let slice_data = original_data.slice(75, 7);
+        let (new_offsets, original_start, length) =
+            reencode_offsets::<i32>(&slice_data.buffers()[0], &slice_data);
+        assert_eq!(
+            vec![0, 3, 6, 9, 12, 15, 18, 21],
+            new_offsets.typed_data::<i32>()
+        );
+        assert_eq!(225, original_start);
+        assert_eq!(21, length);
+    }
+
+    #[test]
+    fn reencode_offsets_when_first_offset_is_zero() {
+        let mut ls = GenericListBuilder::<i32, _>::new(UInt32Builder::new());
+        // ls = [[], [35, 42]
+        ls.append(true);
+        ls.values().append_value(35);
+        ls.values().append_value(42);
+        ls.append(true);
+        let original_list = ls.finish();
+        let original_data = original_list.into_data();
+
+        let slice_data = original_data.slice(1, 1);
+        let (new_offsets, original_start, length) =
+            reencode_offsets::<i32>(&slice_data.buffers()[0], &slice_data);
+        assert_eq!(vec![0, 2], new_offsets.typed_data::<i32>());
+        assert_eq!(0, original_start);
+        assert_eq!(2, length);
+    }
+
     /// Ensure when serde full & sliced versions they are equal to original input.
     /// Also ensure serialized sliced version is significantly smaller than serialized full.
     fn roundtrip_ensure_sliced_smaller(in_batch: RecordBatch, expected_size_factor: usize) {
diff --git a/arrow-ord/src/rank.rs b/arrow-ord/src/rank.rs
index e61cebef3..1b0d2a7e6 100644
--- a/arrow-ord/src/rank.rs
+++ b/arrow-ord/src/rank.rs
@@ -19,7 +19,9 @@
 
 use arrow_array::cast::AsArray;
 use arrow_array::types::*;
-use arrow_array::{downcast_primitive_array, Array, ArrowNativeTypeOp, GenericByteArray};
+use arrow_array::{
+    downcast_primitive_array, Array, ArrowNativeTypeOp, BooleanArray, GenericByteArray,
+};
 use arrow_buffer::NullBuffer;
 use arrow_schema::{ArrowError, DataType, SortOptions};
 use std::cmp::Ordering;
@@ -29,7 +31,11 @@ pub(crate) fn can_rank(data_type: &DataType) -> bool {
     data_type.is_primitive()
         || matches!(
             data_type,
-            DataType::Utf8 | DataType::LargeUtf8 | DataType::Binary | DataType::LargeBinary
+            DataType::Boolean
+                | DataType::Utf8
+                | DataType::LargeUtf8
+                | DataType::Binary
+                | DataType::LargeBinary
         )
 }
 
@@ -49,6 +55,7 @@ pub fn rank(array: &dyn Array, options: Option<SortOptions>) -> Result<Vec<u32>,
     let options = options.unwrap_or_default();
     let ranks = downcast_primitive_array! {
         array => primitive_rank(array.values(), array.nulls(), options),
+        DataType::Boolean => boolean_rank(array.as_boolean(), options),
         DataType::Utf8 => bytes_rank(array.as_bytes::<Utf8Type>(), options),
         DataType::LargeUtf8 => bytes_rank(array.as_bytes::<LargeUtf8Type>(), options),
         DataType::Binary => bytes_rank(array.as_bytes::<BinaryType>(), options),
@@ -135,6 +142,84 @@ where
     out
 }
 
+/// Return the index for the rank when ranking boolean array
+///
+/// The index is calculated as follows:
+/// if is_null is true, the index is 2
+/// if is_null is false and the value is true, the index is 1
+/// otherwise, the index is 0
+///
+/// false is 0 and true is 1 because these are the value when cast to number
+#[inline]
+fn get_boolean_rank_index(value: bool, is_null: bool) -> usize {
+    let is_null_num = is_null as usize;
+    (is_null_num << 1) | (value as usize & !is_null_num)
+}
+
+#[inline(never)]
+fn boolean_rank(array: &BooleanArray, options: SortOptions) -> Vec<u32> {
+    let null_count = array.null_count() as u32;
+    let true_count = array.true_count() as u32;
+    let false_count = array.len() as u32 - null_count - true_count;
+
+    // Rank values for [false, true, null] in that order
+    //
+    // The value for a rank is last value rank + own value count
+    // this means that if we have the following order: `false`, `true` and then `null`
+    // the ranks will be:
+    // - false: false_count
+    // - true: false_count + true_count
+    // - null: false_count + true_count + null_count
+    //
+    // If we have the following order: `null`, `false` and then `true`
+    // the ranks will be:
+    // - false: null_count + false_count
+    // - true: null_count + false_count + true_count
+    // - null: null_count
+    //
+    // You will notice that the last rank is always the total length of the array but we don't use it for readability on how the rank is calculated
+    let ranks_index: [u32; 3] = match (options.descending, options.nulls_first) {
+        // The order is null, true, false
+        (true, true) => [
+            null_count + true_count + false_count,
+            null_count + true_count,
+            null_count,
+        ],
+        // The order is true, false, null
+        (true, false) => [
+            true_count + false_count,
+            true_count,
+            true_count + false_count + null_count,
+        ],
+        // The order is null, false, true
+        (false, true) => [
+            null_count + false_count,
+            null_count + false_count + true_count,
+            null_count,
+        ],
+        // The order is false, true, null
+        (false, false) => [
+            false_count,
+            false_count + true_count,
+            false_count + true_count + null_count,
+        ],
+    };
+
+    match array.nulls().filter(|n| n.null_count() > 0) {
+        Some(n) => array
+            .values()
+            .iter()
+            .zip(n.iter())
+            .map(|(value, is_valid)| ranks_index[get_boolean_rank_index(value, !is_valid)])
+            .collect::<Vec<u32>>(),
+        None => array
+            .values()
+            .iter()
+            .map(|value| ranks_index[value as usize])
+            .collect::<Vec<u32>>(),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -177,6 +262,82 @@ mod tests {
         assert_eq!(res, &[4, 6, 3, 6, 3, 3]);
     }
 
+    #[test]
+    fn test_get_boolean_rank_index() {
+        assert_eq!(get_boolean_rank_index(true, true), 2);
+        assert_eq!(get_boolean_rank_index(false, true), 2);
+        assert_eq!(get_boolean_rank_index(true, false), 1);
+        assert_eq!(get_boolean_rank_index(false, false), 0);
+    }
+
+    #[test]
+    fn test_nullable_booleans() {
+        let descending = SortOptions {
+            descending: true,
+            nulls_first: true,
+        };
+
+        let nulls_last = SortOptions {
+            descending: false,
+            nulls_first: false,
+        };
+
+        let nulls_last_descending = SortOptions {
+            descending: true,
+            nulls_first: false,
+        };
+
+        let a = BooleanArray::from(vec![Some(true), Some(true), None, Some(false), Some(false)]);
+        let res = rank(&a, None).unwrap();
+        assert_eq!(res, &[5, 5, 1, 3, 3]);
+
+        let res = rank(&a, Some(descending)).unwrap();
+        assert_eq!(res, &[3, 3, 1, 5, 5]);
+
+        let res = rank(&a, Some(nulls_last)).unwrap();
+        assert_eq!(res, &[4, 4, 5, 2, 2]);
+
+        let res = rank(&a, Some(nulls_last_descending)).unwrap();
+        assert_eq!(res, &[2, 2, 5, 4, 4]);
+
+        // Test with non-zero null values
+        let nulls = NullBuffer::from(vec![true, true, false, true, true]);
+        let a = BooleanArray::new(vec![true, true, true, false, false].into(), Some(nulls));
+        let res = rank(&a, None).unwrap();
+        assert_eq!(res, &[5, 5, 1, 3, 3]);
+    }
+
+    #[test]
+    fn test_booleans() {
+        let descending = SortOptions {
+            descending: true,
+            nulls_first: true,
+        };
+
+        let nulls_last = SortOptions {
+            descending: false,
+            nulls_first: false,
+        };
+
+        let nulls_last_descending = SortOptions {
+            descending: true,
+            nulls_first: false,
+        };
+
+        let a = BooleanArray::from(vec![true, false, false, false, true]);
+        let res = rank(&a, None).unwrap();
+        assert_eq!(res, &[5, 3, 3, 3, 5]);
+
+        let res = rank(&a, Some(descending)).unwrap();
+        assert_eq!(res, &[2, 5, 5, 5, 2]);
+
+        let res = rank(&a, Some(nulls_last)).unwrap();
+        assert_eq!(res, &[5, 3, 3, 3, 5]);
+
+        let res = rank(&a, Some(nulls_last_descending)).unwrap();
+        assert_eq!(res, &[2, 5, 5, 5, 2]);
+    }
+
     #[test]
     fn test_bytes() {
         let v = vec!["foo", "fo", "bar", "bar"];
diff --git a/arrow-ord/src/sort.rs b/arrow-ord/src/sort.rs
index 51a6659e6..fa5e2b8b2 100644
--- a/arrow-ord/src/sort.rs
+++ b/arrow-ord/src/sort.rs
@@ -785,12 +785,14 @@ impl LexicographicalComparator {
 mod tests {
     use super::*;
     use arrow_array::builder::{
-        FixedSizeListBuilder, Int64Builder, ListBuilder, PrimitiveRunBuilder,
+        BooleanBuilder, FixedSizeListBuilder, GenericListBuilder, Int64Builder, ListBuilder,
+        PrimitiveRunBuilder,
     };
     use arrow_buffer::{i256, NullBuffer};
     use arrow_schema::Field;
     use half::f16;
     use rand::rngs::StdRng;
+    use rand::seq::SliceRandom;
     use rand::{Rng, RngCore, SeedableRng};
 
     fn create_decimal128_array(data: Vec<Option<i128>>) -> Decimal128Array {
@@ -1541,6 +1543,384 @@ mod tests {
         );
     }
 
+    /// Test sort boolean on each permutation of with/without limit and GenericListArray/FixedSizeListArray
+    ///
+    /// The input data must have the same length for all list items so that we can test FixedSizeListArray
+    ///
+    fn test_every_config_sort_boolean_list_arrays(
+        data: Vec<Option<Vec<Option<bool>>>>,
+        options: Option<SortOptions>,
+        expected_data: Vec<Option<Vec<Option<bool>>>>,
+    ) {
+        let first_length = data
+            .iter()
+            .find_map(|x| x.as_ref().map(|x| x.len()))
+            .unwrap_or(0);
+        let first_non_match_length = data
+            .iter()
+            .map(|x| x.as_ref().map(|x| x.len()).unwrap_or(first_length))
+            .position(|x| x != first_length);
+
+        assert_eq!(
+            first_non_match_length, None,
+            "All list items should have the same length {first_length}, input data is invalid"
+        );
+
+        let first_non_match_length = expected_data
+            .iter()
+            .map(|x| x.as_ref().map(|x| x.len()).unwrap_or(first_length))
+            .position(|x| x != first_length);
+
+        assert_eq!(
+            first_non_match_length, None,
+            "All list items should have the same length {first_length}, expected data is invalid"
+        );
+
+        let limit = expected_data.len().saturating_div(2);
+
+        for &with_limit in &[false, true] {
+            let (limit, expected_data) = if with_limit {
+                (
+                    Some(limit),
+                    expected_data.iter().take(limit).cloned().collect(),
+                )
+            } else {
+                (None, expected_data.clone())
+            };
+
+            for &fixed_length in &[None, Some(first_length as i32)] {
+                test_sort_boolean_list_arrays(
+                    data.clone(),
+                    options,
+                    limit,
+                    expected_data.clone(),
+                    fixed_length,
+                );
+            }
+        }
+    }
+
+    fn test_sort_boolean_list_arrays(
+        data: Vec<Option<Vec<Option<bool>>>>,
+        options: Option<SortOptions>,
+        limit: Option<usize>,
+        expected_data: Vec<Option<Vec<Option<bool>>>>,
+        fixed_length: Option<i32>,
+    ) {
+        fn build_fixed_boolean_list_array(
+            data: Vec<Option<Vec<Option<bool>>>>,
+            fixed_length: i32,
+        ) -> ArrayRef {
+            let mut builder = FixedSizeListBuilder::new(
+                BooleanBuilder::with_capacity(fixed_length as usize),
+                fixed_length,
+            );
+            for sublist in data {
+                match sublist {
+                    Some(sublist) => {
+                        builder.values().extend(sublist);
+                        builder.append(true);
+                    }
+                    None => {
+                        builder
+                            .values()
+                            .extend(std::iter::repeat(None).take(fixed_length as usize));
+                        builder.append(false);
+                    }
+                }
+            }
+            Arc::new(builder.finish()) as ArrayRef
+        }
+
+        fn build_generic_boolean_list_array<OffsetSize: OffsetSizeTrait>(
+            data: Vec<Option<Vec<Option<bool>>>>,
+        ) -> ArrayRef {
+            let mut builder = GenericListBuilder::<OffsetSize, _>::new(BooleanBuilder::new());
+            builder.extend(data);
+            Arc::new(builder.finish()) as ArrayRef
+        }
+
+        // for FixedSizedList
+        if let Some(length) = fixed_length {
+            let input = build_fixed_boolean_list_array(data.clone(), length);
+            let sorted = match limit {
+                Some(_) => sort_limit(&(input as ArrayRef), options, limit).unwrap(),
+                _ => sort(&(input as ArrayRef), options).unwrap(),
+            };
+            let expected = build_fixed_boolean_list_array(expected_data.clone(), length);
+
+            assert_eq!(&sorted, &expected);
+        }
+
+        // for List
+        let input = build_generic_boolean_list_array::<i32>(data.clone());
+        let sorted = match limit {
+            Some(_) => sort_limit(&(input as ArrayRef), options, limit).unwrap(),
+            _ => sort(&(input as ArrayRef), options).unwrap(),
+        };
+        let expected = build_generic_boolean_list_array::<i32>(expected_data.clone());
+
+        assert_eq!(&sorted, &expected);
+
+        // for LargeList
+        let input = build_generic_boolean_list_array::<i64>(data.clone());
+        let sorted = match limit {
+            Some(_) => sort_limit(&(input as ArrayRef), options, limit).unwrap(),
+            _ => sort(&(input as ArrayRef), options).unwrap(),
+        };
+        let expected = build_generic_boolean_list_array::<i64>(expected_data.clone());
+
+        assert_eq!(&sorted, &expected);
+    }
+
+    #[test]
+    fn test_sort_list_of_booleans() {
+        // These are all the possible combinations of boolean values
+        // There are 3^3 + 1 = 28 possible combinations (3 values to permutate - [true, false, null] and 1 None value)
+        #[rustfmt::skip]
+        let mut cases = vec![
+            Some(vec![Some(true),  Some(true),  Some(true)]),
+            Some(vec![Some(true),  Some(true),  Some(false)]),
+            Some(vec![Some(true),  Some(true),  None]),
+
+            Some(vec![Some(true),  Some(false), Some(true)]),
+            Some(vec![Some(true),  Some(false), Some(false)]),
+            Some(vec![Some(true),  Some(false), None]),
+
+            Some(vec![Some(true),  None,        Some(true)]),
+            Some(vec![Some(true),  None,        Some(false)]),
+            Some(vec![Some(true),  None,        None]),
+
+            Some(vec![Some(false), Some(true),  Some(true)]),
+            Some(vec![Some(false), Some(true),  Some(false)]),
+            Some(vec![Some(false), Some(true),  None]),
+
+            Some(vec![Some(false), Some(false), Some(true)]),
+            Some(vec![Some(false), Some(false), Some(false)]),
+            Some(vec![Some(false), Some(false), None]),
+
+            Some(vec![Some(false), None,        Some(true)]),
+            Some(vec![Some(false), None,        Some(false)]),
+            Some(vec![Some(false), None,        None]),
+
+            Some(vec![None,        Some(true),  Some(true)]),
+            Some(vec![None,        Some(true),  Some(false)]),
+            Some(vec![None,        Some(true),  None]),
+
+            Some(vec![None,        Some(false), Some(true)]),
+            Some(vec![None,        Some(false), Some(false)]),
+            Some(vec![None,        Some(false), None]),
+
+            Some(vec![None,        None,        Some(true)]),
+            Some(vec![None,        None,        Some(false)]),
+            Some(vec![None,        None,        None]),
+            None,
+        ];
+
+        cases.shuffle(&mut StdRng::seed_from_u64(42));
+
+        // The order is false, true, null
+        #[rustfmt::skip]
+        let expected_descending_false_nulls_first_false = vec![
+            Some(vec![Some(false), Some(false), Some(false)]),
+            Some(vec![Some(false), Some(false), Some(true)]),
+            Some(vec![Some(false), Some(false), None]),
+
+            Some(vec![Some(false), Some(true),  Some(false)]),
+            Some(vec![Some(false), Some(true),  Some(true)]),
+            Some(vec![Some(false), Some(true),  None]),
+
+            Some(vec![Some(false), None,        Some(false)]),
+            Some(vec![Some(false), None,        Some(true)]),
+            Some(vec![Some(false), None,        None]),
+
+            Some(vec![Some(true),  Some(false), Some(false)]),
+            Some(vec![Some(true),  Some(false), Some(true)]),
+            Some(vec![Some(true),  Some(false), None]),
+
+            Some(vec![Some(true),  Some(true),  Some(false)]),
+            Some(vec![Some(true),  Some(true),  Some(true)]),
+            Some(vec![Some(true),  Some(true),  None]),
+
+            Some(vec![Some(true),  None,        Some(false)]),
+            Some(vec![Some(true),  None,        Some(true)]),
+            Some(vec![Some(true),  None,        None]),
+
+            Some(vec![None,        Some(false), Some(false)]),
+            Some(vec![None,        Some(false), Some(true)]),
+            Some(vec![None,        Some(false), None]),
+
+            Some(vec![None,        Some(true),  Some(false)]),
+            Some(vec![None,        Some(true),  Some(true)]),
+            Some(vec![None,        Some(true),  None]),
+
+            Some(vec![None,        None,        Some(false)]),
+            Some(vec![None,        None,        Some(true)]),
+            Some(vec![None,        None,        None]),
+            None,
+        ];
+        test_every_config_sort_boolean_list_arrays(
+            cases.clone(),
+            Some(SortOptions {
+                descending: false,
+                nulls_first: false,
+            }),
+            expected_descending_false_nulls_first_false,
+        );
+
+        // The order is null, false, true
+        #[rustfmt::skip]
+        let expected_descending_false_nulls_first_true = vec![
+            None,
+
+            Some(vec![None,        None,        None]),
+            Some(vec![None,        None,        Some(false)]),
+            Some(vec![None,        None,        Some(true)]),
+
+            Some(vec![None,        Some(false), None]),
+            Some(vec![None,        Some(false), Some(false)]),
+            Some(vec![None,        Some(false), Some(true)]),
+
+            Some(vec![None,        Some(true),  None]),
+            Some(vec![None,        Some(true),  Some(false)]),
+            Some(vec![None,        Some(true),  Some(true)]),
+
+            Some(vec![Some(false), None,        None]),
+            Some(vec![Some(false), None,        Some(false)]),
+            Some(vec![Some(false), None,        Some(true)]),
+
+            Some(vec![Some(false), Some(false), None]),
+            Some(vec![Some(false), Some(false), Some(false)]),
+            Some(vec![Some(false), Some(false), Some(true)]),
+
+            Some(vec![Some(false), Some(true),  None]),
+            Some(vec![Some(false), Some(true),  Some(false)]),
+            Some(vec![Some(false), Some(true),  Some(true)]),
+
+            Some(vec![Some(true),  None,        None]),
+            Some(vec![Some(true),  None,        Some(false)]),
+            Some(vec![Some(true),  None,        Some(true)]),
+
+            Some(vec![Some(true),  Some(false), None]),
+            Some(vec![Some(true),  Some(false), Some(false)]),
+            Some(vec![Some(true),  Some(false), Some(true)]),
+
+            Some(vec![Some(true),  Some(true),  None]),
+            Some(vec![Some(true),  Some(true),  Some(false)]),
+            Some(vec![Some(true),  Some(true),  Some(true)]),
+        ];
+
+        test_every_config_sort_boolean_list_arrays(
+            cases.clone(),
+            Some(SortOptions {
+                descending: false,
+                nulls_first: true,
+            }),
+            expected_descending_false_nulls_first_true,
+        );
+
+        // The order is true, false, null
+        #[rustfmt::skip]
+        let expected_descending_true_nulls_first_false = vec![
+            Some(vec![Some(true),  Some(true),  Some(true)]),
+            Some(vec![Some(true),  Some(true),  Some(false)]),
+            Some(vec![Some(true),  Some(true),  None]),
+
+            Some(vec![Some(true),  Some(false), Some(true)]),
+            Some(vec![Some(true),  Some(false), Some(false)]),
+            Some(vec![Some(true),  Some(false), None]),
+
+            Some(vec![Some(true),  None,        Some(true)]),
+            Some(vec![Some(true),  None,        Some(false)]),
+            Some(vec![Some(true),  None,        None]),
+
+            Some(vec![Some(false), Some(true),  Some(true)]),
+            Some(vec![Some(false), Some(true),  Some(false)]),
+            Some(vec![Some(false), Some(true),  None]),
+
+            Some(vec![Some(false), Some(false), Some(true)]),
+            Some(vec![Some(false), Some(false), Some(false)]),
+            Some(vec![Some(false), Some(false), None]),
+
+            Some(vec![Some(false), None,        Some(true)]),
+            Some(vec![Some(false), None,        Some(false)]),
+            Some(vec![Some(false), None,        None]),
+
+            Some(vec![None,        Some(true),  Some(true)]),
+            Some(vec![None,        Some(true),  Some(false)]),
+            Some(vec![None,        Some(true),  None]),
+
+            Some(vec![None,        Some(false), Some(true)]),
+            Some(vec![None,        Some(false), Some(false)]),
+            Some(vec![None,        Some(false), None]),
+
+            Some(vec![None,        None,        Some(true)]),
+            Some(vec![None,        None,        Some(false)]),
+            Some(vec![None,        None,        None]),
+
+            None,
+        ];
+        test_every_config_sort_boolean_list_arrays(
+            cases.clone(),
+            Some(SortOptions {
+                descending: true,
+                nulls_first: false,
+            }),
+            expected_descending_true_nulls_first_false,
+        );
+
+        // The order is null, true, false
+        #[rustfmt::skip]
+        let expected_descending_true_nulls_first_true = vec![
+            None,
+
+            Some(vec![None,        None,        None]),
+            Some(vec![None,        None,        Some(true)]),
+            Some(vec![None,        None,        Some(false)]),
+
+            Some(vec![None,        Some(true),  None]),
+            Some(vec![None,        Some(true),  Some(true)]),
+            Some(vec![None,        Some(true),  Some(false)]),
+
+            Some(vec![None,        Some(false), None]),
+            Some(vec![None,        Some(false), Some(true)]),
+            Some(vec![None,        Some(false), Some(false)]),
+
+            Some(vec![Some(true),  None,        None]),
+            Some(vec![Some(true),  None,        Some(true)]),
+            Some(vec![Some(true),  None,        Some(false)]),
+
+            Some(vec![Some(true),  Some(true),  None]),
+            Some(vec![Some(true),  Some(true),  Some(true)]),
+            Some(vec![Some(true),  Some(true),  Some(false)]),
+
+            Some(vec![Some(true),  Some(false), None]),
+            Some(vec![Some(true),  Some(false), Some(true)]),
+            Some(vec![Some(true),  Some(false), Some(false)]),
+
+            Some(vec![Some(false), None,        None]),
+            Some(vec![Some(false), None,        Some(true)]),
+            Some(vec![Some(false), None,        Some(false)]),
+
+            Some(vec![Some(false), Some(true),  None]),
+            Some(vec![Some(false), Some(true),  Some(true)]),
+            Some(vec![Some(false), Some(true),  Some(false)]),
+
+            Some(vec![Some(false), Some(false), None]),
+            Some(vec![Some(false), Some(false), Some(true)]),
+            Some(vec![Some(false), Some(false), Some(false)]),
+        ];
+        // Testing with limit false and fixed_length None
+        test_every_config_sort_boolean_list_arrays(
+            cases.clone(),
+            Some(SortOptions {
+                descending: true,
+                nulls_first: true,
+            }),
+            expected_descending_true_nulls_first_true,
+        );
+    }
+
     #[test]
     fn test_sort_indices_decimal128() {
         // decimal default
diff --git a/arrow-schema/src/datatype.rs b/arrow-schema/src/datatype.rs
index 7cd53b13c..5c9073c4e 100644
--- a/arrow-schema/src/datatype.rs
+++ b/arrow-schema/src/datatype.rs
@@ -197,6 +197,17 @@ pub enum DataType {
     /// DataType::Timestamp(TimeUnit::Second, Some("string".to_string().into()));
     /// ```
     ///
+    /// # Timezone representation
+    /// ----------------------------
+    /// It is possible to use either the timezone string representation, such as "UTC", or the absolute time zone offset "+00:00".
+    /// For timezones with fixed offsets, such as "UTC" or "JST", the offset representation is recommended, as it is more explicit and less ambiguous.
+    ///
+    /// Most arrow-rs functionalities use the absolute offset representation,
+    /// such as [`PrimitiveArray::with_timezone_utc`] that applies a
+    /// UTC timezone to timestamp arrays.
+    ///
+    /// [`PrimitiveArray::with_timezone_utc`]: https://docs.rs/arrow/latest/arrow/array/struct.PrimitiveArray.html#method.with_timezone_utc
+    ///
     /// Timezone string parsing
     /// -----------------------
     /// When feature `chrono-tz` is not enabled, allowed timezone strings are fixed offsets of the form "+09:00", "-09" or "+0930".
@@ -234,7 +245,7 @@ pub enum DataType {
     ///
     /// # Recommendation
     ///
-    /// Users should prefer [`DataType::Date32`] to cleanly represent the number
+    /// Users should prefer [`Date32`] to cleanly represent the number
     /// of days, or one of the Timestamp variants to include time as part of the
     /// representation, depending on their use case.
     ///
@@ -242,6 +253,7 @@ pub enum DataType {
     ///
     /// For more details, see [#5288](https://github.com/apache/arrow-rs/issues/5288).
     ///
+    /// [`Date32`]: Self::Date32
     /// [Schema.fbs]: https://github.com/apache/arrow/blob/main/format/Schema.fbs
     Date64,
     /// A signed 32-bit time representing the elapsed time since midnight in the unit of `TimeUnit`.
@@ -271,10 +283,12 @@ pub enum DataType {
     LargeBinary,
     /// Opaque binary data of variable length.
     ///
-    /// Logically the same as [`Self::Binary`], but the internal representation uses a view
+    /// Logically the same as [`Binary`], but the internal representation uses a view
     /// struct that contains the string length and either the string's entire data
     /// inline (for small strings) or an inlined prefix, an index of another buffer,
     /// and an offset pointing to a slice in that buffer (for non-small strings).
+    ///
+    /// [`Binary`]: Self::Binary
     BinaryView,
     /// A variable-length string in Unicode with UTF-8 encoding.
     ///
@@ -288,10 +302,12 @@ pub enum DataType {
     LargeUtf8,
     /// A variable-length string in Unicode with UTF-8 encoding
     ///
-    /// Logically the same as [`Self::Utf8`], but the internal representation uses a view
+    /// Logically the same as [`Utf8`], but the internal representation uses a view
     /// struct that contains the string length and either the string's entire data
     /// inline (for small strings) or an inlined prefix, an index of another buffer,
     /// and an offset pointing to a slice in that buffer (for non-small strings).
+    ///
+    /// [`Utf8`]: Self::Utf8
     Utf8View,
     /// A list of some logical data type with variable length.
     ///
@@ -300,11 +316,12 @@ pub enum DataType {
 
     /// (NOT YET FULLY SUPPORTED)  A list of some logical data type with variable length.
     ///
+    /// Logically the same as [`List`], but the internal representation differs in how child
+    /// data is referenced, allowing flexibility in how data is layed out.
+    ///
     /// Note this data type is not yet fully supported. Using it with arrow APIs may result in `panic`s.
     ///
-    /// The ListView layout is defined by three buffers:
-    /// a validity bitmap, an offsets buffer, and an additional sizes buffer.
-    /// Sizes and offsets are both 32 bits for this type
+    /// [`List`]: Self::List
     ListView(FieldRef),
     /// A list of some logical data type with fixed length.
     FixedSizeList(FieldRef, i32),
@@ -315,11 +332,12 @@ pub enum DataType {
 
     /// (NOT YET FULLY SUPPORTED)  A list of some logical data type with variable length and 64-bit offsets.
     ///
+    /// Logically the same as [`LargeList`], but the internal representation differs in how child
+    /// data is referenced, allowing flexibility in how data is layed out.
+    ///
     /// Note this data type is not yet fully supported. Using it with arrow APIs may result in `panic`s.
     ///
-    /// The LargeListView layout is defined by three buffers:
-    /// a validity bitmap, an offsets buffer, and an additional sizes buffer.
-    /// Sizes and offsets are both 64 bits for this type
+    /// [`LargeList`]: Self::LargeList
     LargeListView(FieldRef),
     /// A nested datatype that contains a number of sub-fields.
     Struct(Fields),
@@ -552,16 +570,21 @@ impl DataType {
         matches!(self, Int16 | Int32 | Int64)
     }
 
-    /// Returns true if this type is nested (List, FixedSizeList, LargeList, Struct, Union,
+    /// Returns true if this type is nested (List, FixedSizeList, LargeList, ListView. LargeListView, Struct, Union,
     /// or Map), or a dictionary of a nested type
     #[inline]
     pub fn is_nested(&self) -> bool {
         use DataType::*;
         match self {
             Dictionary(_, v) => DataType::is_nested(v.as_ref()),
-            List(_) | FixedSizeList(_, _) | LargeList(_) | Struct(_) | Union(_, _) | Map(_, _) => {
-                true
-            }
+            List(_)
+            | FixedSizeList(_, _)
+            | LargeList(_)
+            | ListView(_)
+            | LargeListView(_)
+            | Struct(_)
+            | Union(_, _)
+            | Map(_, _) => true,
             _ => false,
         }
     }
@@ -578,7 +601,9 @@ impl DataType {
     pub fn equals_datatype(&self, other: &DataType) -> bool {
         match (&self, other) {
             (DataType::List(a), DataType::List(b))
-            | (DataType::LargeList(a), DataType::LargeList(b)) => {
+            | (DataType::LargeList(a), DataType::LargeList(b))
+            | (DataType::ListView(a), DataType::ListView(b))
+            | (DataType::LargeListView(a), DataType::LargeListView(b)) => {
                 a.is_nullable() == b.is_nullable() && a.data_type().equals_datatype(b.data_type())
             }
             (DataType::FixedSizeList(a, a_size), DataType::FixedSizeList(b, b_size)) => {
@@ -725,7 +750,9 @@ impl DataType {
     pub fn contains(&self, other: &DataType) -> bool {
         match (self, other) {
             (DataType::List(f1), DataType::List(f2))
-            | (DataType::LargeList(f1), DataType::LargeList(f2)) => f1.contains(f2),
+            | (DataType::LargeList(f1), DataType::LargeList(f2))
+            | (DataType::ListView(f1), DataType::ListView(f2))
+            | (DataType::LargeListView(f1), DataType::LargeListView(f2)) => f1.contains(f2),
             (DataType::FixedSizeList(f1, s1), DataType::FixedSizeList(f2, s2)) => {
                 s1 == s2 && f1.contains(f2)
             }
@@ -1006,11 +1033,16 @@ mod tests {
     #[test]
     fn test_nested() {
         let list = DataType::List(Arc::new(Field::new("foo", DataType::Utf8, true)));
+        let list_view = DataType::ListView(Arc::new(Field::new("foo", DataType::Utf8, true)));
+        let large_list_view =
+            DataType::LargeListView(Arc::new(Field::new("foo", DataType::Utf8, true)));
 
         assert!(!DataType::is_nested(&DataType::Boolean));
         assert!(!DataType::is_nested(&DataType::Int32));
         assert!(!DataType::is_nested(&DataType::Utf8));
         assert!(DataType::is_nested(&list));
+        assert!(DataType::is_nested(&list_view));
+        assert!(DataType::is_nested(&large_list_view));
 
         assert!(!DataType::is_nested(&DataType::Dictionary(
             Box::new(DataType::Int32),
diff --git a/arrow-schema/src/schema.rs b/arrow-schema/src/schema.rs
index 6c79da53f..9affd4162 100644
--- a/arrow-schema/src/schema.rs
+++ b/arrow-schema/src/schema.rs
@@ -22,7 +22,7 @@ use std::sync::Arc;
 
 use crate::error::ArrowError;
 use crate::field::Field;
-use crate::{FieldRef, Fields};
+use crate::{DataType, FieldRef, Fields};
 
 /// A builder to facilitate building a [`Schema`] from iteratively from [`FieldRef`]
 #[derive(Debug, Default)]
@@ -418,6 +418,89 @@ impl Schema {
         &self.metadata
     }
 
+    /// Normalize a [`Schema`] into a flat table.
+    ///
+    /// Nested [`Field`]s will generate names separated by `separator`, up to a depth of `max_level`
+    /// (unlimited if `None`).
+    ///
+    /// e.g. given a [`Schema`]:
+    ///
+    /// ```text
+    ///     "foo": StructArray<"bar": Utf8>
+    /// ```
+    ///
+    /// A separator of `"."` would generate a batch with the schema:
+    ///
+    /// ```text
+    ///     "foo.bar": Utf8
+    /// ```
+    ///
+    /// Note that giving a depth of `Some(0)` to `max_level` is the same as passing in `None`;
+    /// it will be treated as unlimited.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use std::sync::Arc;
+    /// # use arrow_schema::{DataType, Field, Fields, Schema};
+    /// let schema = Schema::new(vec![
+    ///     Field::new(
+    ///         "a",
+    ///         DataType::Struct(Fields::from(vec![
+    ///             Arc::new(Field::new("animals", DataType::Utf8, true)),
+    ///             Arc::new(Field::new("n_legs", DataType::Int64, true)),
+    ///         ])),
+    ///         false,
+    ///     ),
+    /// ])
+    /// .normalize(".", None)
+    /// .expect("valid normalization");
+    /// let expected = Schema::new(vec![
+    ///     Field::new("a.animals", DataType::Utf8, true),
+    ///     Field::new("a.n_legs", DataType::Int64, true),
+    /// ]);
+    /// assert_eq!(schema, expected);
+    /// ```
+    pub fn normalize(&self, separator: &str, max_level: Option<usize>) -> Result<Self, ArrowError> {
+        let max_level = match max_level.unwrap_or(usize::MAX) {
+            0 => usize::MAX,
+            val => val,
+        };
+        let mut stack: Vec<(usize, Vec<&str>, &FieldRef)> = self
+            .fields()
+            .iter()
+            .rev()
+            .map(|f| {
+                let name_vec: Vec<&str> = vec![f.name()];
+                (0, name_vec, f)
+            })
+            .collect();
+        let mut fields: Vec<FieldRef> = Vec::new();
+
+        while let Some((depth, name, field_ref)) = stack.pop() {
+            match field_ref.data_type() {
+                DataType::Struct(ff) if depth < max_level => {
+                    // Need to zip these in reverse to maintain original order
+                    for fff in ff.into_iter().rev() {
+                        let mut name = name.clone();
+                        name.push(separator);
+                        name.push(fff.name());
+                        stack.push((depth + 1, name, fff))
+                    }
+                }
+                _ => {
+                    let updated_field = Field::new(
+                        name.concat(),
+                        field_ref.data_type().clone(),
+                        field_ref.is_nullable(),
+                    );
+                    fields.push(Arc::new(updated_field));
+                }
+            }
+        }
+        Ok(Schema::new(fields))
+    }
+
     /// Look up a column by name and return a immutable reference to the column along with
     /// its index.
     pub fn column_with_name(&self, name: &str) -> Option<(usize, &Field)> {
@@ -679,6 +762,410 @@ mod tests {
         schema.index_of("nickname").unwrap();
     }
 
+    #[test]
+    fn normalize_simple() {
+        let schema = Schema::new(vec![
+            Field::new(
+                "a",
+                DataType::Struct(Fields::from(vec![
+                    Arc::new(Field::new("animals", DataType::Utf8, true)),
+                    Arc::new(Field::new("n_legs", DataType::Int64, true)),
+                    Arc::new(Field::new("year", DataType::Int64, true)),
+                ])),
+                false,
+            ),
+            Field::new("month", DataType::Int64, true),
+        ])
+        .normalize(".", Some(0))
+        .expect("valid normalization");
+
+        let expected = Schema::new(vec![
+            Field::new("a.animals", DataType::Utf8, true),
+            Field::new("a.n_legs", DataType::Int64, true),
+            Field::new("a.year", DataType::Int64, true),
+            Field::new("month", DataType::Int64, true),
+        ]);
+
+        assert_eq!(schema, expected);
+
+        // Check that 0, None have the same result
+        let schema = Schema::new(vec![
+            Field::new(
+                "a",
+                DataType::Struct(Fields::from(vec![
+                    Arc::new(Field::new("animals", DataType::Utf8, true)),
+                    Arc::new(Field::new("n_legs", DataType::Int64, true)),
+                    Arc::new(Field::new("year", DataType::Int64, true)),
+                ])),
+                false,
+            ),
+            Field::new("month", DataType::Int64, true),
+        ])
+        .normalize(".", None)
+        .expect("valid normalization");
+
+        assert_eq!(schema, expected);
+    }
+
+    #[test]
+    fn normalize_nested() {
+        let a = Arc::new(Field::new("a", DataType::Utf8, true));
+        let b = Arc::new(Field::new("b", DataType::Int64, false));
+        let c = Arc::new(Field::new("c", DataType::Int64, true));
+
+        let d = Arc::new(Field::new("d", DataType::Utf8, true));
+        let e = Arc::new(Field::new("e", DataType::Int64, false));
+        let f = Arc::new(Field::new("f", DataType::Int64, true));
+
+        let one = Arc::new(Field::new(
+            "1",
+            DataType::Struct(Fields::from(vec![a.clone(), b.clone(), c.clone()])),
+            false,
+        ));
+        let two = Arc::new(Field::new(
+            "2",
+            DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+            true,
+        ));
+
+        let exclamation = Arc::new(Field::new(
+            "!",
+            DataType::Struct(Fields::from(vec![one, two])),
+            false,
+        ));
+
+        let normalize_all = Schema::new(vec![exclamation.clone()])
+            .normalize(".", Some(0))
+            .expect("valid normalization");
+
+        let expected = Schema::new(vec![
+            Field::new("!.1.a", DataType::Utf8, true),
+            Field::new("!.1.b", DataType::Int64, false),
+            Field::new("!.1.c", DataType::Int64, true),
+            Field::new("!.2.d", DataType::Utf8, true),
+            Field::new("!.2.e", DataType::Int64, false),
+            Field::new("!.2.f", DataType::Int64, true),
+        ]);
+
+        assert_eq!(normalize_all, expected);
+
+        let normalize_depth_one = Schema::new(vec![exclamation])
+            .normalize(".", Some(1))
+            .expect("valid normalization");
+
+        let expected = Schema::new(vec![
+            Field::new("!.1", DataType::Struct(Fields::from(vec![a, b, c])), false),
+            Field::new("!.2", DataType::Struct(Fields::from(vec![d, e, f])), true),
+        ]);
+
+        assert_eq!(normalize_depth_one, expected);
+    }
+
+    #[test]
+    fn normalize_list() {
+        // Only the Struct type field should be unwrapped
+        let a = Arc::new(Field::new("a", DataType::Utf8, true));
+        let b = Arc::new(Field::new("b", DataType::Int64, false));
+        let c = Arc::new(Field::new("c", DataType::Int64, true));
+        let d = Arc::new(Field::new("d", DataType::Utf8, true));
+        let e = Arc::new(Field::new("e", DataType::Int64, false));
+        let f = Arc::new(Field::new("f", DataType::Int64, true));
+
+        let one = Arc::new(Field::new(
+            "1",
+            DataType::Struct(Fields::from(vec![a.clone(), b.clone(), c.clone()])),
+            true,
+        ));
+
+        let two = Arc::new(Field::new(
+            "2",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                true,
+            ))),
+            false,
+        ));
+
+        let exclamation = Arc::new(Field::new(
+            "!",
+            DataType::Struct(Fields::from(vec![one.clone(), two.clone()])),
+            false,
+        ));
+
+        let normalize_all = Schema::new(vec![exclamation.clone()])
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        // List shouldn't be affected
+        let expected = Schema::new(vec![
+            Field::new("!.1.a", DataType::Utf8, true),
+            Field::new("!.1.b", DataType::Int64, false),
+            Field::new("!.1.c", DataType::Int64, true),
+            Field::new(
+                "!.2",
+                DataType::List(Arc::new(Field::new_list_field(
+                    DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                    true,
+                ))),
+                false,
+            ),
+        ]);
+
+        assert_eq!(normalize_all, expected);
+        assert_eq!(normalize_all.fields().len(), 4);
+
+        // FixedSizeList
+        let two = Arc::new(Field::new(
+            "2",
+            DataType::FixedSizeList(
+                Arc::new(Field::new_fixed_size_list(
+                    "3",
+                    Arc::new(Field::new_list_field(
+                        DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                        true,
+                    )),
+                    1,
+                    true,
+                )),
+                1,
+            ),
+            false,
+        ));
+
+        let exclamation = Arc::new(Field::new(
+            "!",
+            DataType::Struct(Fields::from(vec![one.clone(), two])),
+            false,
+        ));
+
+        let normalize_all = Schema::new(vec![exclamation.clone()])
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        // FixedSizeList shouldn't be affected
+        let expected = Schema::new(vec![
+            Field::new("!.1.a", DataType::Utf8, true),
+            Field::new("!.1.b", DataType::Int64, false),
+            Field::new("!.1.c", DataType::Int64, true),
+            Field::new(
+                "!.2",
+                DataType::FixedSizeList(
+                    Arc::new(Field::new_fixed_size_list(
+                        "3",
+                        Arc::new(Field::new_list_field(
+                            DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                            true,
+                        )),
+                        1,
+                        true,
+                    )),
+                    1,
+                ),
+                false,
+            ),
+        ]);
+
+        assert_eq!(normalize_all, expected);
+        assert_eq!(normalize_all.fields().len(), 4);
+
+        // LargeList
+        let two = Arc::new(Field::new(
+            "2",
+            DataType::FixedSizeList(
+                Arc::new(Field::new_large_list(
+                    "3",
+                    Arc::new(Field::new_list_field(
+                        DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                        true,
+                    )),
+                    true,
+                )),
+                1,
+            ),
+            false,
+        ));
+
+        let exclamation = Arc::new(Field::new(
+            "!",
+            DataType::Struct(Fields::from(vec![one.clone(), two])),
+            false,
+        ));
+
+        let normalize_all = Schema::new(vec![exclamation.clone()])
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        // LargeList shouldn't be affected
+        let expected = Schema::new(vec![
+            Field::new("!.1.a", DataType::Utf8, true),
+            Field::new("!.1.b", DataType::Int64, false),
+            Field::new("!.1.c", DataType::Int64, true),
+            Field::new(
+                "!.2",
+                DataType::FixedSizeList(
+                    Arc::new(Field::new_large_list(
+                        "3",
+                        Arc::new(Field::new_list_field(
+                            DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                            true,
+                        )),
+                        true,
+                    )),
+                    1,
+                ),
+                false,
+            ),
+        ]);
+
+        assert_eq!(normalize_all, expected);
+        assert_eq!(normalize_all.fields().len(), 4);
+    }
+
+    #[test]
+    fn normalize_deep_nested() {
+        // No unwrapping expected
+        let a = Arc::new(Field::new("a", DataType::Utf8, true));
+        let b = Arc::new(Field::new("b", DataType::Int64, false));
+        let c = Arc::new(Field::new("c", DataType::Int64, true));
+        let d = Arc::new(Field::new("d", DataType::Utf8, true));
+        let e = Arc::new(Field::new("e", DataType::Int64, false));
+        let f = Arc::new(Field::new("f", DataType::Int64, true));
+
+        let one = Arc::new(Field::new(
+            "1",
+            DataType::Struct(Fields::from(vec![a.clone(), b.clone(), c.clone()])),
+            true,
+        ));
+
+        let two = Arc::new(Field::new(
+            "2",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![d.clone(), e.clone(), f.clone()])),
+                true,
+            ))),
+            false,
+        ));
+
+        let l10 = Arc::new(Field::new(
+            "l10",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![one, two])),
+                true,
+            ))),
+            false,
+        ));
+
+        let l9 = Arc::new(Field::new(
+            "l9",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l10])),
+                true,
+            ))),
+            false,
+        ));
+
+        let l8 = Arc::new(Field::new(
+            "l8",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l9])),
+                true,
+            ))),
+            false,
+        ));
+        let l7 = Arc::new(Field::new(
+            "l7",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l8])),
+                true,
+            ))),
+            false,
+        ));
+        let l6 = Arc::new(Field::new(
+            "l6",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l7])),
+                true,
+            ))),
+            false,
+        ));
+        let l5 = Arc::new(Field::new(
+            "l5",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l6])),
+                true,
+            ))),
+            false,
+        ));
+        let l4 = Arc::new(Field::new(
+            "l4",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l5])),
+                true,
+            ))),
+            false,
+        ));
+        let l3 = Arc::new(Field::new(
+            "l3",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l4])),
+                true,
+            ))),
+            false,
+        ));
+        let l2 = Arc::new(Field::new(
+            "l2",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l3])),
+                true,
+            ))),
+            false,
+        ));
+        let l1 = Arc::new(Field::new(
+            "l1",
+            DataType::List(Arc::new(Field::new_list_field(
+                DataType::Struct(Fields::from(vec![l2])),
+                true,
+            ))),
+            false,
+        ));
+
+        let normalize_all = Schema::new(vec![l1])
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        assert_eq!(normalize_all.fields().len(), 1);
+    }
+
+    #[test]
+    fn normalize_dictionary() {
+        let a = Arc::new(Field::new("a", DataType::Utf8, true));
+        let b = Arc::new(Field::new("b", DataType::Int64, false));
+
+        let one = Arc::new(Field::new(
+            "1",
+            DataType::Dictionary(
+                Box::new(DataType::Int32),
+                Box::new(DataType::Struct(Fields::from(vec![a.clone(), b.clone()]))),
+            ),
+            false,
+        ));
+
+        let normalize_all = Schema::new(vec![one.clone()])
+            .normalize(".", None)
+            .expect("valid normalization");
+
+        let expected = Schema::new(vec![Field::new(
+            "1",
+            DataType::Dictionary(
+                Box::new(DataType::Int32),
+                Box::new(DataType::Struct(Fields::from(vec![a.clone(), b.clone()]))),
+            ),
+            false,
+        )]);
+
+        assert_eq!(normalize_all, expected);
+    }
+
     #[test]
     #[should_panic(
         expected = "Unable to get field named \\\"nickname\\\". Valid fields: [\\\"first_name\\\", \\\"last_name\\\", \\\"address\\\", \\\"interests\\\"]"
diff --git a/arrow-select/src/interleave.rs b/arrow-select/src/interleave.rs
index 4a47017b7..3557bda8f 100644
--- a/arrow-select/src/interleave.rs
+++ b/arrow-select/src/interleave.rs
@@ -24,7 +24,9 @@ use arrow_array::types::*;
 use arrow_array::*;
 use arrow_buffer::{ArrowNativeType, MutableBuffer, NullBuffer, NullBufferBuilder, OffsetBuffer};
 use arrow_data::transform::MutableArrayData;
+use arrow_data::ByteView;
 use arrow_schema::{ArrowError, DataType};
+use std::collections::HashMap;
 use std::sync::Arc;
 
 macro_rules! primitive_helper {
@@ -97,6 +99,8 @@ pub fn interleave(
         DataType::LargeUtf8 => interleave_bytes::<LargeUtf8Type>(values, indices),
         DataType::Binary => interleave_bytes::<BinaryType>(values, indices),
         DataType::LargeBinary => interleave_bytes::<LargeBinaryType>(values, indices),
+        DataType::BinaryView => interleave_views::<BinaryViewType>(values, indices),
+        DataType::Utf8View => interleave_views::<StringViewType>(values, indices),
         DataType::Dictionary(k, _) => downcast_integer! {
             k.as_ref() => (dict_helper, values, indices),
             _ => unreachable!("illegal dictionary key type {k}")
@@ -231,6 +235,41 @@ fn interleave_dictionaries<K: ArrowDictionaryKeyType>(
     Ok(Arc::new(array))
 }
 
+fn interleave_views<T: ByteViewType>(
+    values: &[&dyn Array],
+    indices: &[(usize, usize)],
+) -> Result<ArrayRef, ArrowError> {
+    let interleaved = Interleave::<'_, GenericByteViewArray<T>>::new(values, indices);
+    let mut views_builder = BufferBuilder::new(indices.len());
+    let mut buffers = Vec::new();
+
+    // (input array_index, input buffer_index) -> output buffer_index
+    let mut buffer_lookup: HashMap<(usize, u32), u32> = HashMap::new();
+    for (array_idx, value_idx) in indices {
+        let array = interleaved.arrays[*array_idx];
+        let raw_view = array.views().get(*value_idx).unwrap();
+        let view_len = *raw_view as u32;
+        if view_len <= 12 {
+            views_builder.append(*raw_view);
+            continue;
+        }
+        // value is big enough to be in a variadic buffer
+        let view = ByteView::from(*raw_view);
+        let new_buffer_idx: &mut u32 = buffer_lookup
+            .entry((*array_idx, view.buffer_index))
+            .or_insert_with(|| {
+                buffers.push(array.data_buffers()[view.buffer_index as usize].clone());
+                (buffers.len() - 1) as u32
+            });
+        views_builder.append(view.with_buffer_index(*new_buffer_idx).into());
+    }
+
+    let array = unsafe {
+        GenericByteViewArray::<T>::new_unchecked(views_builder.into(), buffers, interleaved.nulls)
+    };
+    Ok(Arc::new(array))
+}
+
 /// Fallback implementation of interleave using [`MutableArrayData`]
 fn interleave_fallback(
     values: &[&dyn Array],
@@ -461,4 +500,209 @@ mod tests {
             DictionaryArray::<Int32Type>::from_iter(vec![Some("0"), Some("1"), Some("2"), None]);
         assert_eq!(array.as_ref(), &expected)
     }
+
+    #[test]
+    fn test_interleave_views() {
+        let values = StringArray::from_iter_values([
+            "hello",
+            "world_long_string_not_inlined",
+            "foo",
+            "bar",
+            "baz",
+        ]);
+        let view_a = StringViewArray::from(&values);
+
+        let values = StringArray::from_iter_values([
+            "test",
+            "data",
+            "more_long_string_not_inlined",
+            "views",
+            "here",
+        ]);
+        let view_b = StringViewArray::from(&values);
+
+        let indices = &[
+            (0, 2), // "foo"
+            (1, 0), // "test"
+            (0, 4), // "baz"
+            (1, 3), // "views"
+            (0, 1), // "world_long_string_not_inlined"
+        ];
+
+        // Test specialized implementation
+        let values = interleave(&[&view_a, &view_b], indices).unwrap();
+        let result = values.as_string_view();
+        assert_eq!(result.data_buffers().len(), 1);
+
+        let fallback = interleave_fallback(&[&view_a, &view_b], indices).unwrap();
+        let fallback_result = fallback.as_string_view();
+        // note that fallback_result has 2 buffers, but only one long enough string to warrant a buffer
+        assert_eq!(fallback_result.data_buffers().len(), 2);
+
+        // Convert to strings for easier assertion
+        let collected: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
+
+        let fallback_collected: Vec<_> = fallback_result
+            .iter()
+            .map(|x| x.map(|s| s.to_string()))
+            .collect();
+
+        assert_eq!(&collected, &fallback_collected);
+
+        assert_eq!(
+            &collected,
+            &[
+                Some("foo".to_string()),
+                Some("test".to_string()),
+                Some("baz".to_string()),
+                Some("views".to_string()),
+                Some("world_long_string_not_inlined".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn test_interleave_views_with_nulls() {
+        let values = StringArray::from_iter([
+            Some("hello"),
+            None,
+            Some("foo_long_string_not_inlined"),
+            Some("bar"),
+            None,
+        ]);
+        let view_a = StringViewArray::from(&values);
+
+        let values = StringArray::from_iter([
+            Some("test"),
+            Some("data_long_string_not_inlined"),
+            None,
+            None,
+            Some("here"),
+        ]);
+        let view_b = StringViewArray::from(&values);
+
+        let indices = &[
+            (0, 1), // null
+            (1, 2), // null
+            (0, 2), // "foo_long_string_not_inlined"
+            (1, 3), // null
+            (0, 4), // null
+        ];
+
+        // Test specialized implementation
+        let values = interleave(&[&view_a, &view_b], indices).unwrap();
+        let result = values.as_string_view();
+        assert_eq!(result.data_buffers().len(), 1);
+
+        let fallback = interleave_fallback(&[&view_a, &view_b], indices).unwrap();
+        let fallback_result = fallback.as_string_view();
+
+        // Convert to strings for easier assertion
+        let collected: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
+
+        let fallback_collected: Vec<_> = fallback_result
+            .iter()
+            .map(|x| x.map(|s| s.to_string()))
+            .collect();
+
+        assert_eq!(&collected, &fallback_collected);
+
+        assert_eq!(
+            &collected,
+            &[
+                None,
+                None,
+                Some("foo_long_string_not_inlined".to_string()),
+                None,
+                None,
+            ]
+        );
+    }
+
+    #[test]
+    fn test_interleave_views_multiple_buffers() {
+        let str1 = "very_long_string_from_first_buffer".as_bytes();
+        let str2 = "very_long_string_from_second_buffer".as_bytes();
+        let buffer1 = str1.to_vec().into();
+        let buffer2 = str2.to_vec().into();
+
+        let view1 = ByteView::new(str1.len() as u32, &str1[..4])
+            .with_buffer_index(0)
+            .with_offset(0)
+            .as_u128();
+        let view2 = ByteView::new(str2.len() as u32, &str2[..4])
+            .with_buffer_index(1)
+            .with_offset(0)
+            .as_u128();
+        let view_a =
+            StringViewArray::try_new(vec![view1, view2].into(), vec![buffer1, buffer2], None)
+                .unwrap();
+
+        let str3 = "another_very_long_string_buffer_three".as_bytes();
+        let str4 = "different_long_string_in_buffer_four".as_bytes();
+        let buffer3 = str3.to_vec().into();
+        let buffer4 = str4.to_vec().into();
+
+        let view3 = ByteView::new(str3.len() as u32, &str3[..4])
+            .with_buffer_index(0)
+            .with_offset(0)
+            .as_u128();
+        let view4 = ByteView::new(str4.len() as u32, &str4[..4])
+            .with_buffer_index(1)
+            .with_offset(0)
+            .as_u128();
+        let view_b =
+            StringViewArray::try_new(vec![view3, view4].into(), vec![buffer3, buffer4], None)
+                .unwrap();
+
+        let indices = &[
+            (0, 0), // String from first buffer of array A
+            (1, 0), // String from first buffer of array B
+            (0, 1), // String from second buffer of array A
+            (1, 1), // String from second buffer of array B
+            (0, 0), // String from first buffer of array A again
+            (1, 1), // String from second buffer of array B again
+        ];
+
+        // Test interleave
+        let values = interleave(&[&view_a, &view_b], indices).unwrap();
+        let result = values.as_string_view();
+
+        assert_eq!(
+            result.data_buffers().len(),
+            4,
+            "Expected four buffers (two from each input array)"
+        );
+
+        let result_strings: Vec<_> = result.iter().map(|x| x.map(|s| s.to_string())).collect();
+        assert_eq!(
+            result_strings,
+            vec![
+                Some("very_long_string_from_first_buffer".to_string()),
+                Some("another_very_long_string_buffer_three".to_string()),
+                Some("very_long_string_from_second_buffer".to_string()),
+                Some("different_long_string_in_buffer_four".to_string()),
+                Some("very_long_string_from_first_buffer".to_string()),
+                Some("different_long_string_in_buffer_four".to_string()),
+            ]
+        );
+
+        let views = result.views();
+        let buffer_indices: Vec<_> = views
+            .iter()
+            .map(|raw_view| ByteView::from(*raw_view).buffer_index)
+            .collect();
+
+        assert_eq!(
+            buffer_indices,
+            vec![
+                0, // First buffer from array A
+                1, // First buffer from array B
+                2, // Second buffer from array A
+                3, // Second buffer from array B
+                0, // First buffer from array A (reused)
+                3, // Second buffer from array B (reused)
+            ]
+        );
+    }
 }
diff --git a/arrow-string/src/binary_like.rs b/arrow-string/src/binary_like.rs
new file mode 100644
index 000000000..8f8641edb
--- /dev/null
+++ b/arrow-string/src/binary_like.rs
@@ -0,0 +1,166 @@
+// 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.
+
+//! Provide SQL's CONTAINS, STARTS_WITH, ENDS_WITH operators for Arrow's binary arrays
+
+use crate::binary_predicate::BinaryPredicate;
+
+use arrow_array::cast::AsArray;
+use arrow_array::*;
+use arrow_schema::*;
+use arrow_select::take::take;
+
+#[derive(Debug)]
+pub(crate) enum Op {
+    Contains,
+    StartsWith,
+    EndsWith,
+}
+
+impl TryFrom<crate::like::Op> for Op {
+    type Error = ArrowError;
+
+    fn try_from(value: crate::like::Op) -> Result<Self, Self::Error> {
+        match value {
+            crate::like::Op::Contains => Ok(Op::Contains),
+            crate::like::Op::StartsWith => Ok(Op::StartsWith),
+            crate::like::Op::EndsWith => Ok(Op::EndsWith),
+            _ => Err(ArrowError::InvalidArgumentError(format!(
+                "Invalid binary operation: {value}"
+            ))),
+        }
+    }
+}
+
+impl std::fmt::Display for Op {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Op::Contains => write!(f, "CONTAINS"),
+            Op::StartsWith => write!(f, "STARTS_WITH"),
+            Op::EndsWith => write!(f, "ENDS_WITH"),
+        }
+    }
+}
+
+pub(crate) fn binary_apply<'a, 'i, T: BinaryArrayType<'a> + 'a>(
+    op: Op,
+    l: T,
+    l_s: bool,
+    l_v: Option<&'a dyn AnyDictionaryArray>,
+    r: T,
+    r_s: bool,
+    r_v: Option<&'a dyn AnyDictionaryArray>,
+) -> Result<BooleanArray, ArrowError> {
+    let l_len = l_v.map(|l| l.len()).unwrap_or(l.len());
+    if r_s {
+        let idx = match r_v {
+            Some(dict) if dict.null_count() != 0 => return Ok(BooleanArray::new_null(l_len)),
+            Some(dict) => dict.normalized_keys()[0],
+            None => 0,
+        };
+        if r.is_null(idx) {
+            return Ok(BooleanArray::new_null(l_len));
+        }
+        op_scalar::<T>(op, l, l_v, r.value(idx))
+    } else {
+        match (l_s, l_v, r_v) {
+            (true, None, None) => {
+                let v = l.is_valid(0).then(|| l.value(0));
+                op_binary(op, std::iter::repeat(v), r.iter())
+            }
+            (true, Some(l_v), None) => {
+                let idx = l_v.is_valid(0).then(|| l_v.normalized_keys()[0]);
+                let v = idx.and_then(|idx| l.is_valid(idx).then(|| l.value(idx)));
+                op_binary(op, std::iter::repeat(v), r.iter())
+            }
+            (true, None, Some(r_v)) => {
+                let v = l.is_valid(0).then(|| l.value(0));
+                op_binary(op, std::iter::repeat(v), vectored_iter(r, r_v))
+            }
+            (true, Some(l_v), Some(r_v)) => {
+                let idx = l_v.is_valid(0).then(|| l_v.normalized_keys()[0]);
+                let v = idx.and_then(|idx| l.is_valid(idx).then(|| l.value(idx)));
+                op_binary(op, std::iter::repeat(v), vectored_iter(r, r_v))
+            }
+            (false, None, None) => op_binary(op, l.iter(), r.iter()),
+            (false, Some(l_v), None) => op_binary(op, vectored_iter(l, l_v), r.iter()),
+            (false, None, Some(r_v)) => op_binary(op, l.iter(), vectored_iter(r, r_v)),
+            (false, Some(l_v), Some(r_v)) => {
+                op_binary(op, vectored_iter(l, l_v), vectored_iter(r, r_v))
+            }
+        }
+    }
+}
+
+#[inline(never)]
+fn op_scalar<'a, T: BinaryArrayType<'a>>(
+    op: Op,
+    l: T,
+    l_v: Option<&dyn AnyDictionaryArray>,
+    r: &[u8],
+) -> Result<BooleanArray, ArrowError> {
+    let r = match op {
+        Op::Contains => BinaryPredicate::contains(r).evaluate_array(l, false),
+        Op::StartsWith => BinaryPredicate::StartsWith(r).evaluate_array(l, false),
+        Op::EndsWith => BinaryPredicate::EndsWith(r).evaluate_array(l, false),
+    };
+
+    Ok(match l_v {
+        Some(v) => take(&r, v.keys(), None)?.as_boolean().clone(),
+        None => r,
+    })
+}
+
+fn vectored_iter<'a, T: BinaryArrayType<'a> + 'a>(
+    a: T,
+    a_v: &'a dyn AnyDictionaryArray,
+) -> impl Iterator<Item = Option<&'a [u8]>> + 'a {
+    let nulls = a_v.nulls();
+    let keys = a_v.normalized_keys();
+    keys.into_iter().enumerate().map(move |(idx, key)| {
+        if nulls.map(|n| n.is_null(idx)).unwrap_or_default() || a.is_null(key) {
+            return None;
+        }
+        Some(a.value(key))
+    })
+}
+
+#[inline(never)]
+fn op_binary<'a>(
+    op: Op,
+    l: impl Iterator<Item = Option<&'a [u8]>>,
+    r: impl Iterator<Item = Option<&'a [u8]>>,
+) -> Result<BooleanArray, ArrowError> {
+    match op {
+        Op::Contains => Ok(l
+            .zip(r)
+            .map(|(l, r)| Some(bytes_contains(l?, r?)))
+            .collect()),
+        Op::StartsWith => Ok(l
+            .zip(r)
+            .map(|(l, r)| Some(BinaryPredicate::StartsWith(r?).evaluate(l?)))
+            .collect()),
+        Op::EndsWith => Ok(l
+            .zip(r)
+            .map(|(l, r)| Some(BinaryPredicate::EndsWith(r?).evaluate(l?)))
+            .collect()),
+    }
+}
+
+fn bytes_contains(haystack: &[u8], needle: &[u8]) -> bool {
+    memchr::memmem::find(haystack, needle).is_some()
+}
diff --git a/arrow-string/src/binary_predicate.rs b/arrow-string/src/binary_predicate.rs
new file mode 100644
index 000000000..b14e3c255
--- /dev/null
+++ b/arrow-string/src/binary_predicate.rs
@@ -0,0 +1,178 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow_array::{Array, ArrayAccessor, BinaryViewArray, BooleanArray};
+use arrow_buffer::BooleanBuffer;
+use memchr::memmem::Finder;
+use std::iter::zip;
+
+/// A binary based predicate
+pub enum BinaryPredicate<'a> {
+    Contains(Finder<'a>),
+    StartsWith(&'a [u8]),
+    EndsWith(&'a [u8]),
+}
+
+impl<'a> BinaryPredicate<'a> {
+    pub fn contains(needle: &'a [u8]) -> Self {
+        Self::Contains(Finder::new(needle))
+    }
+
+    /// Evaluate this predicate against the given haystack
+    pub fn evaluate(&self, haystack: &[u8]) -> bool {
+        match self {
+            Self::Contains(finder) => finder.find(haystack).is_some(),
+            Self::StartsWith(v) => starts_with(haystack, v, equals_kernel),
+            Self::EndsWith(v) => ends_with(haystack, v, equals_kernel),
+        }
+    }
+
+    /// Evaluate this predicate against the elements of `array`
+    ///
+    /// If `negate` is true the result of the predicate will be negated
+    #[inline(never)]
+    pub fn evaluate_array<'i, T>(&self, array: T, negate: bool) -> BooleanArray
+    where
+        T: ArrayAccessor<Item = &'i [u8]>,
+    {
+        match self {
+            Self::Contains(finder) => BooleanArray::from_unary(array, |haystack| {
+                finder.find(haystack).is_some() != negate
+            }),
+            Self::StartsWith(v) => {
+                if let Some(view_array) = array.as_any().downcast_ref::<BinaryViewArray>() {
+                    let nulls = view_array.logical_nulls();
+                    let values = BooleanBuffer::from(
+                        view_array
+                            .prefix_bytes_iter(v.len())
+                            .map(|haystack| equals_bytes(haystack, v, equals_kernel) != negate)
+                            .collect::<Vec<_>>(),
+                    );
+                    BooleanArray::new(values, nulls)
+                } else {
+                    BooleanArray::from_unary(array, |haystack| {
+                        starts_with(haystack, v, equals_kernel) != negate
+                    })
+                }
+            }
+            Self::EndsWith(v) => {
+                if let Some(view_array) = array.as_any().downcast_ref::<BinaryViewArray>() {
+                    let nulls = view_array.logical_nulls();
+                    let values = BooleanBuffer::from(
+                        view_array
+                            .suffix_bytes_iter(v.len())
+                            .map(|haystack| equals_bytes(haystack, v, equals_kernel) != negate)
+                            .collect::<Vec<_>>(),
+                    );
+                    BooleanArray::new(values, nulls)
+                } else {
+                    BooleanArray::from_unary(array, |haystack| {
+                        ends_with(haystack, v, equals_kernel) != negate
+                    })
+                }
+            }
+        }
+    }
+}
+
+fn equals_bytes(lhs: &[u8], rhs: &[u8], byte_eq_kernel: impl Fn((&u8, &u8)) -> bool) -> bool {
+    lhs.len() == rhs.len() && zip(lhs, rhs).all(byte_eq_kernel)
+}
+
+/// This is faster than `[u8]::starts_with` for small slices.
+/// See <https://github.com/apache/arrow-rs/issues/6107> for more details.
+fn starts_with(
+    haystack: &[u8],
+    needle: &[u8],
+    byte_eq_kernel: impl Fn((&u8, &u8)) -> bool,
+) -> bool {
+    if needle.len() > haystack.len() {
+        false
+    } else {
+        zip(haystack, needle).all(byte_eq_kernel)
+    }
+}
+/// This is faster than `[u8]::ends_with` for small slices.
+/// See <https://github.com/apache/arrow-rs/issues/6107> for more details.
+fn ends_with(haystack: &[u8], needle: &[u8], byte_eq_kernel: impl Fn((&u8, &u8)) -> bool) -> bool {
+    if needle.len() > haystack.len() {
+        false
+    } else {
+        zip(haystack.iter().rev(), needle.iter().rev()).all(byte_eq_kernel)
+    }
+}
+
+fn equals_kernel((n, h): (&u8, &u8)) -> bool {
+    n == h
+}
+
+#[cfg(test)]
+mod tests {
+    use super::BinaryPredicate;
+
+    #[test]
+    fn test_contains() {
+        assert!(BinaryPredicate::contains(b"hay").evaluate(b"haystack"));
+        assert!(BinaryPredicate::contains(b"haystack").evaluate(b"haystack"));
+        assert!(BinaryPredicate::contains(b"h").evaluate(b"haystack"));
+        assert!(BinaryPredicate::contains(b"k").evaluate(b"haystack"));
+        assert!(BinaryPredicate::contains(b"stack").evaluate(b"haystack"));
+        assert!(BinaryPredicate::contains(b"sta").evaluate(b"haystack"));
+        assert!(BinaryPredicate::contains(b"stack").evaluate(b"hay\0stack"));
+        assert!(BinaryPredicate::contains(b"\0s").evaluate(b"hay\0stack"));
+        assert!(BinaryPredicate::contains(b"\0").evaluate(b"hay\0stack"));
+        assert!(BinaryPredicate::contains(b"a").evaluate(b"a"));
+        // not matching
+        assert!(!BinaryPredicate::contains(b"hy").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::contains(b"stackx").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::contains(b"x").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::contains(b"haystack haystack").evaluate(b"haystack"));
+    }
+
+    #[test]
+    fn test_starts_with() {
+        assert!(BinaryPredicate::StartsWith(b"hay").evaluate(b"haystack"));
+        assert!(BinaryPredicate::StartsWith(b"h\0ay").evaluate(b"h\0aystack"));
+        assert!(BinaryPredicate::StartsWith(b"haystack").evaluate(b"haystack"));
+        assert!(BinaryPredicate::StartsWith(b"ha").evaluate(b"haystack"));
+        assert!(BinaryPredicate::StartsWith(b"h").evaluate(b"haystack"));
+        assert!(BinaryPredicate::StartsWith(b"").evaluate(b"haystack"));
+
+        assert!(!BinaryPredicate::StartsWith(b"stack").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::StartsWith(b"haystacks").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::StartsWith(b"HAY").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::StartsWith(b"h\0ay").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::StartsWith(b"hay").evaluate(b"h\0aystack"));
+    }
+
+    #[test]
+    fn test_ends_with() {
+        assert!(BinaryPredicate::EndsWith(b"stack").evaluate(b"haystack"));
+        assert!(BinaryPredicate::EndsWith(b"st\0ack").evaluate(b"hayst\0ack"));
+        assert!(BinaryPredicate::EndsWith(b"haystack").evaluate(b"haystack"));
+        assert!(BinaryPredicate::EndsWith(b"ck").evaluate(b"haystack"));
+        assert!(BinaryPredicate::EndsWith(b"k").evaluate(b"haystack"));
+        assert!(BinaryPredicate::EndsWith(b"").evaluate(b"haystack"));
+
+        assert!(!BinaryPredicate::EndsWith(b"hay").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::EndsWith(b"STACK").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::EndsWith(b"haystacks").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::EndsWith(b"xhaystack").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::EndsWith(b"st\0ack").evaluate(b"haystack"));
+        assert!(!BinaryPredicate::EndsWith(b"stack").evaluate(b"hayst\0ack"));
+    }
+}
diff --git a/arrow-string/src/lib.rs b/arrow-string/src/lib.rs
index 87e23a7b2..d199c38bf 100644
--- a/arrow-string/src/lib.rs
+++ b/arrow-string/src/lib.rs
@@ -18,6 +18,8 @@
 #![warn(missing_docs)]
 //! Arrow string kernels
 
+mod binary_like;
+mod binary_predicate;
 pub mod concat_elements;
 pub mod length;
 pub mod like;
diff --git a/arrow-string/src/like.rs b/arrow-string/src/like.rs
index e30e09146..91ba94fd6 100644
--- a/arrow-string/src/like.rs
+++ b/arrow-string/src/like.rs
@@ -26,10 +26,11 @@ use arrow_select::take::take;
 
 use std::sync::Arc;
 
+use crate::binary_like::binary_apply;
 pub use arrow_array::StringArrayType;
 
 #[derive(Debug)]
-enum Op {
+pub(crate) enum Op {
     Like(bool),
     ILike(bool),
     Contains,
@@ -53,16 +54,22 @@ impl std::fmt::Display for Op {
 
 /// Perform SQL `left LIKE right`
 ///
+/// # Supported DataTypes
+///
+/// `left` and `right` must be the same type, and one of
+/// - Utf8
+/// - LargeUtf8
+/// - Utf8View
+///
 /// There are two wildcards supported with the LIKE operator:
 ///
 /// 1. `%` - The percent sign represents zero, one, or multiple characters
 /// 2. `_` - The underscore represents a single character
 ///
-/// For example:
+/// Example
 /// ```
 /// # use arrow_array::{StringArray, BooleanArray};
 /// # use arrow_string::like::like;
-/// #
 /// let strings = StringArray::from(vec!["Arrow", "Arrow", "Arrow", "Ar"]);
 /// let patterns = StringArray::from(vec!["A%", "B%", "A.", "A_"]);
 ///
@@ -75,39 +82,107 @@ pub fn like(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowEr
 
 /// Perform SQL `left ILIKE right`
 ///
-/// This is a case-insensitive version of [`like`]
-///
-/// Note: this only implements loose matching as defined by the Unicode standard. For example,
-/// the `ﬀ` ligature is not equivalent to `FF` and `ß` is not equivalent to `SS`
+/// # Notes
+/// - This is a case-insensitive version of [`like`]
+/// - See the documentation on [`like`] for more details
+/// - Implements loose matching as defined by the Unicode standard. For example,
+///   the `ﬀ` ligature is not equivalent to `FF` and `ß` is not equivalent to `SS`
 pub fn ilike(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowError> {
     like_op(Op::ILike(false), left, right)
 }
 
 /// Perform SQL `left NOT LIKE right`
 ///
-/// See the documentation on [`like`] for more details
+/// # Notes
+/// - This is a negative of [`like`]
+/// - See the documentation on [`like`] for more details
 pub fn nlike(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowError> {
     like_op(Op::Like(true), left, right)
 }
 
 /// Perform SQL `left NOT ILIKE right`
 ///
-/// See the documentation on [`ilike`] for more details
+/// # Notes
+/// - This is a negative of [`like`]
+/// - See the documentation on [`ilike`] for more details
 pub fn nilike(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowError> {
     like_op(Op::ILike(true), left, right)
 }
 
 /// Perform SQL `STARTSWITH(left, right)`
+///
+/// # Supported DataTypes
+///
+/// `left` and `right` must be the same type, and one of
+/// - Utf8
+/// - LargeUtf8
+/// - Utf8View
+/// - Binary
+/// - LargeBinary
+/// - BinaryView
+///
+/// # Example
+/// ```
+/// # use arrow_array::{StringArray, BooleanArray};
+/// # use arrow_string::like::{like, starts_with};
+/// let strings = StringArray::from(vec!["arrow-rs", "arrow-rs", "arrow-rs", "Parquet"]);
+/// let patterns = StringArray::from(vec!["arr", "arrow", "arrow-cpp", "p"]);
+///
+/// let result = starts_with(&strings, &patterns).unwrap();
+/// assert_eq!(result, BooleanArray::from(vec![true, true, false, false]));
+/// ```
 pub fn starts_with(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowError> {
     like_op(Op::StartsWith, left, right)
 }
 
 /// Perform SQL `ENDSWITH(left, right)`
+///
+/// # Supported DataTypes
+///
+/// `left` and `right` must be the same type, and one of
+/// - Utf8
+/// - LargeUtf8
+/// - Utf8View
+/// - Binary
+/// - LargeBinary
+/// - BinaryView
+///
+/// # Example
+/// ```
+/// # use arrow_array::{StringArray, BooleanArray};
+/// # use arrow_string::like::{ends_with, like, starts_with};
+/// let strings = StringArray::from(vec!["arrow-rs", "arrow-rs",  "Parquet"]);
+/// let patterns = StringArray::from(vec!["arr", "-rs", "t"]);
+///
+/// let result = ends_with(&strings, &patterns).unwrap();
+/// assert_eq!(result, BooleanArray::from(vec![false, true, true]));
+/// ```
 pub fn ends_with(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowError> {
     like_op(Op::EndsWith, left, right)
 }
 
 /// Perform SQL `CONTAINS(left, right)`
+///
+/// # Supported DataTypes
+///
+/// `left` and `right` must be the same type, and one of
+/// - Utf8
+/// - LargeUtf8
+/// - Utf8View
+/// - Binary
+/// - LargeBinary
+/// - BinaryView
+///
+/// # Example
+/// ```
+/// # use arrow_array::{StringArray, BooleanArray};
+/// # use arrow_string::like::{contains, like, starts_with};
+/// let strings = StringArray::from(vec!["arrow-rs", "arrow-rs", "arrow-rs", "Parquet"]);
+/// let patterns = StringArray::from(vec!["arr", "-rs", "arrow-cpp", "X"]);
+///
+/// let result = contains(&strings, &patterns).unwrap();
+/// assert_eq!(result, BooleanArray::from(vec![true, true, false, false]));
+/// ```
 pub fn contains(left: &dyn Datum, right: &dyn Datum) -> Result<BooleanArray, ArrowError> {
     like_op(Op::Contains, left, right)
 }
@@ -132,13 +207,25 @@ fn like_op(op: Op, lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, Arr
     let r = r_v.map(|x| x.values().as_ref()).unwrap_or(r);
 
     match (l.data_type(), r.data_type()) {
-        (Utf8, Utf8) => {
-            apply::<&GenericStringArray<i32>>(op, l.as_string(), l_s, l_v, r.as_string(), r_s, r_v)
-        }
-        (LargeUtf8, LargeUtf8) => {
-            apply::<&GenericStringArray<i64>>(op, l.as_string(), l_s, l_v, r.as_string(), r_s, r_v)
-        }
-        (Utf8View, Utf8View) => apply::<&StringViewArray>(
+        (Utf8, Utf8) => string_apply::<&GenericStringArray<i32>>(
+            op,
+            l.as_string(),
+            l_s,
+            l_v,
+            r.as_string(),
+            r_s,
+            r_v,
+        ),
+        (LargeUtf8, LargeUtf8) => string_apply::<&GenericStringArray<i64>>(
+            op,
+            l.as_string(),
+            l_s,
+            l_v,
+            r.as_string(),
+            r_s,
+            r_v,
+        ),
+        (Utf8View, Utf8View) => string_apply::<&StringViewArray>(
             op,
             l.as_string_view(),
             l_s,
@@ -147,13 +234,40 @@ fn like_op(op: Op, lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, Arr
             r_s,
             r_v,
         ),
+        (Binary, Binary) => binary_apply::<&GenericBinaryArray<i32>>(
+            op.try_into()?,
+            l.as_binary(),
+            l_s,
+            l_v,
+            r.as_binary(),
+            r_s,
+            r_v,
+        ),
+        (LargeBinary, LargeBinary) => binary_apply::<&GenericBinaryArray<i64>>(
+            op.try_into()?,
+            l.as_binary(),
+            l_s,
+            l_v,
+            r.as_binary(),
+            r_s,
+            r_v,
+        ),
+        (BinaryView, BinaryView) => binary_apply::<&BinaryViewArray>(
+            op.try_into()?,
+            l.as_binary_view(),
+            l_s,
+            l_v,
+            r.as_binary_view(),
+            r_s,
+            r_v,
+        ),
         (l_t, r_t) => Err(ArrowError::InvalidArgumentError(format!(
-            "Invalid string operation: {l_t} {op} {r_t}"
+            "Invalid string/binary operation: {l_t} {op} {r_t}"
         ))),
     }
 }
 
-fn apply<'a, T: StringArrayType<'a> + 'a>(
+fn string_apply<'a, T: StringArrayType<'a> + 'a>(
     op: Op,
     l: T,
     l_s: bool,
@@ -398,9 +512,29 @@ legacy_kernels!(
 #[allow(deprecated)]
 mod tests {
     use super::*;
-    use arrow_array::types::Int8Type;
+    use arrow_array::builder::BinaryDictionaryBuilder;
+    use arrow_array::types::{ArrowDictionaryKeyType, Int8Type};
     use std::iter::zip;
 
+    fn convert_binary_iterator_to_binary_dictionary<
+        'a,
+        K: ArrowDictionaryKeyType,
+        I: IntoIterator<Item = &'a [u8]>,
+    >(
+        iter: I,
+    ) -> DictionaryArray<K> {
+        let it = iter.into_iter();
+        let (lower, _) = it.size_hint();
+        let mut builder = BinaryDictionaryBuilder::with_capacity(lower, 256, 1024);
+        it.for_each(|i| {
+            builder
+                .append(i)
+                .expect("Unable to append a value to a dictionary array.");
+        });
+
+        builder.finish()
+    }
+
     /// Applying `op(left, right)`, both sides are arrays
     /// The macro tests four types of array implementations:
     /// - `StringArray`
@@ -436,6 +570,61 @@ mod tests {
         };
     }
 
+    /// Applying `op(left, right)`, both sides are arrays
+    /// The macro tests four types of array implementations:
+    /// - `StringArray`
+    /// - `LargeStringArray`
+    /// - `StringViewArray`
+    /// - `DictionaryArray`
+    macro_rules! test_utf8_and_binary {
+        ($test_name:ident, $left:expr, $right:expr, $op:expr, $expected:expr) => {
+            #[test]
+            fn $test_name() {
+                let expected = BooleanArray::from($expected);
+
+                let left = StringArray::from($left);
+                let right = StringArray::from($right);
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+
+                let left = LargeStringArray::from($left);
+                let right = LargeStringArray::from($right);
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+
+                let left = StringViewArray::from($left);
+                let right = StringViewArray::from($right);
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+
+                let left: DictionaryArray<Int8Type> = $left.into_iter().collect();
+                let right: DictionaryArray<Int8Type> = $right.into_iter().collect();
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+
+                let left_binary = $left.iter().map(|x| x.as_bytes()).collect::<Vec<&[u8]>>();
+                let right_binary = $right.iter().map(|x| x.as_bytes()).collect::<Vec<&[u8]>>();
+
+                let left = BinaryArray::from(left_binary.clone());
+                let right = BinaryArray::from(right_binary.clone());
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+
+                let left = LargeBinaryArray::from(left_binary.clone());
+                let right = LargeBinaryArray::from(right_binary.clone());
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+
+                let left: DictionaryArray<Int8Type> =
+                    convert_binary_iterator_to_binary_dictionary(left_binary);
+                let right: DictionaryArray<Int8Type> =
+                    convert_binary_iterator_to_binary_dictionary(right_binary);
+                let res = $op(&left, &right).unwrap();
+                assert_eq!(res, expected);
+            }
+        };
+    }
+
     /// Applying `op(left, right)`, left side is array, right side is scalar
     /// The macro tests four types of array implementations:
     /// - `StringArray`
@@ -471,6 +660,61 @@ mod tests {
         };
     }
 
+    /// Applying `op(left, right)`, left side is array, right side is scalar
+    /// The macro tests four types of array implementations:
+    /// - `StringArray`
+    /// - `LargeStringArray`
+    /// - `StringViewArray`
+    /// - `DictionaryArray`
+    macro_rules! test_utf8_and_binary_scalar {
+        ($test_name:ident, $left:expr, $right:expr, $op:expr, $expected:expr) => {
+            #[test]
+            fn $test_name() {
+                let expected = BooleanArray::from($expected);
+
+                let left = StringArray::from($left);
+                let right = StringArray::from_iter_values([$right]);
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+
+                let left = LargeStringArray::from($left);
+                let right = LargeStringArray::from_iter_values([$right]);
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+
+                let left = StringViewArray::from($left);
+                let right = StringViewArray::from_iter_values([$right]);
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+
+                let left: DictionaryArray<Int8Type> = $left.into_iter().collect();
+                let right: DictionaryArray<Int8Type> = [$right].into_iter().collect();
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+
+                let left_binary = $left.iter().map(|x| x.as_bytes()).collect::<Vec<&[u8]>>();
+                let right_binary = $right.as_bytes();
+
+                let left = BinaryArray::from(left_binary.clone());
+                let right = BinaryArray::from_iter_values([right_binary]);
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+
+                let left = LargeBinaryArray::from(left_binary.clone());
+                let right = LargeBinaryArray::from_iter_values([right_binary]);
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+
+                let left: DictionaryArray<Int8Type> =
+                    convert_binary_iterator_to_binary_dictionary(left_binary);
+                let right: DictionaryArray<Int8Type> =
+                    convert_binary_iterator_to_binary_dictionary([right_binary]);
+                let res = $op(&left, &Scalar::new(&right)).unwrap();
+                assert_eq!(res, expected);
+            }
+        };
+    }
+
     test_utf8!(
         test_utf8_array_like,
         vec![
@@ -547,8 +791,8 @@ mod tests {
 
     // Replicates `test_utf8_array_like_scalar_start` `test_utf8_array_like_scalar_dyn_start` to
     // demonstrate that `SQL STARTSWITH` works as expected.
-    test_utf8_scalar!(
-        test_utf8_array_starts_with_scalar_start,
+    test_utf8_and_binary_scalar!(
+        test_utf8_and_binary_array_starts_with_scalar_start,
         vec![
             "arrow",
             "parrow",
@@ -561,6 +805,23 @@ mod tests {
         vec![true, false, true, false, true]
     );
 
+    test_utf8_and_binary!(
+        test_utf8_and_binary_array_starts_with,
+        vec![
+            "arrow",
+            "arrow_long_string_more than 12 bytes",
+            "arrow",
+            "arrow",
+            "arrow",
+            "arrows",
+            "arrow",
+            "arrow"
+        ],
+        vec!["arrow", "ar%", "row", "foo", "arr", "arrow_", "arrow_", ".*"],
+        starts_with,
+        vec![true, false, false, false, true, false, false, false]
+    );
+
     test_utf8_scalar!(
         test_utf8_array_like_scalar_end,
         vec![
@@ -577,8 +838,8 @@ mod tests {
 
     // Replicates `test_utf8_array_like_scalar_end` `test_utf8_array_like_scalar_dyn_end` to
     // demonstrate that `SQL ENDSWITH` works as expected.
-    test_utf8_scalar!(
-        test_utf8_array_ends_with_scalar_end,
+    test_utf8_and_binary_scalar!(
+        test_utf8_and_binary_array_ends_with_scalar_end,
         vec![
             "arrow",
             "parrow",
@@ -591,6 +852,23 @@ mod tests {
         vec![true, true, false, false, false]
     );
 
+    test_utf8_and_binary!(
+        test_utf8_and_binary_array_ends_with,
+        vec![
+            "arrow",
+            "arrow_long_string_more than 12 bytes",
+            "arrow",
+            "arrow",
+            "arrow",
+            "arrows",
+            "arrow",
+            "arrow"
+        ],
+        vec!["arrow", "ar%", "row", "foo", "arr", "arrow_", "arrow_", ".*"],
+        ends_with,
+        vec![true, false, true, false, false, false, false, false]
+    );
+
     test_utf8_scalar!(
         test_utf8_array_like_scalar_equals,
         vec![
@@ -924,8 +1202,8 @@ mod tests {
     // demonstrate that `SQL CONTAINS` works as expected.
     //
     // NOTE: 5 of the values were changed because the original used a case insensitive `ilike`.
-    test_utf8_scalar!(
-        test_utf8_array_contains_unicode_contains,
+    test_utf8_and_binary_scalar!(
+        test_utf8_and_binary_array_contains_unicode_contains,
         vec![
             "sdlkdfFkoßsdfs",
             "sdlkdFFkoSSdggs", // Original was case insensitive "sdlkdfFkoSSdggs"
diff --git a/arrow-string/src/predicate.rs b/arrow-string/src/predicate.rs
index 8bbfe65ba..d785c22b4 100644
--- a/arrow-string/src/predicate.rs
+++ b/arrow-string/src/predicate.rs
@@ -24,7 +24,7 @@ use regex::{Regex, RegexBuilder};
 use std::iter::zip;
 
 /// A string based predicate
-pub enum Predicate<'a> {
+pub(crate) enum Predicate<'a> {
     Eq(&'a str),
     Contains(Finder<'a>),
     StartsWith(&'a str),
@@ -42,7 +42,7 @@ pub enum Predicate<'a> {
 
 impl<'a> Predicate<'a> {
     /// Create a predicate for the given like pattern
-    pub fn like(pattern: &'a str) -> Result<Self, ArrowError> {
+    pub(crate) fn like(pattern: &'a str) -> Result<Self, ArrowError> {
         if !contains_like_pattern(pattern) {
             Ok(Self::Eq(pattern))
         } else if pattern.ends_with('%') && !contains_like_pattern(&pattern[..pattern.len() - 1]) {
@@ -59,12 +59,12 @@ impl<'a> Predicate<'a> {
         }
     }
 
-    pub fn contains(needle: &'a str) -> Self {
+    pub(crate) fn contains(needle: &'a str) -> Self {
         Self::Contains(Finder::new(needle.as_bytes()))
     }
 
     /// Create a predicate for the given ilike pattern
-    pub fn ilike(pattern: &'a str, is_ascii: bool) -> Result<Self, ArrowError> {
+    pub(crate) fn ilike(pattern: &'a str, is_ascii: bool) -> Result<Self, ArrowError> {
         if is_ascii && pattern.is_ascii() {
             if !contains_like_pattern(pattern) {
                 return Ok(Self::IEqAscii(pattern));
@@ -81,7 +81,7 @@ impl<'a> Predicate<'a> {
     }
 
     /// Evaluate this predicate against the given haystack
-    pub fn evaluate(&self, haystack: &str) -> bool {
+    pub(crate) fn evaluate(&self, haystack: &str) -> bool {
         match self {
             Predicate::Eq(v) => *v == haystack,
             Predicate::IEqAscii(v) => haystack.eq_ignore_ascii_case(v),
@@ -100,7 +100,7 @@ impl<'a> Predicate<'a> {
     ///
     /// If `negate` is true the result of the predicate will be negated
     #[inline(never)]
-    pub fn evaluate_array<'i, T>(&self, array: T, negate: bool) -> BooleanArray
+    pub(crate) fn evaluate_array<'i, T>(&self, array: T, negate: bool) -> BooleanArray
     where
         T: ArrayAccessor<Item = &'i str>,
     {
diff --git a/arrow/Cargo.toml b/arrow/Cargo.toml
index 76119ec4a..1b01dcd25 100644
--- a/arrow/Cargo.toml
+++ b/arrow/Cargo.toml
@@ -87,6 +87,9 @@ criterion = { version = "0.5", default-features = false }
 half = { version = "2.1", default-features = false }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
 serde = { version = "1.0", default-features = false, features = ["derive"] }
+# used in examples
+memmap2 = "0.9.3"
+bytes = "1.9"
 
 [build-dependencies]
 
@@ -105,6 +108,11 @@ name = "read_csv_infer_schema"
 required-features = ["prettyprint", "csv"]
 path = "./examples/read_csv_infer_schema.rs"
 
+[[example]]
+name = "zero_copy_ipc"
+required-features = ["prettyprint"]
+path = "examples/zero_copy_ipc.rs"
+
 [[bench]]
 name = "aggregate_kernels"
 harness = false
diff --git a/arrow/examples/README.md b/arrow/examples/README.md
index 87aa6ee04..9d896cd11 100644
--- a/arrow/examples/README.md
+++ b/arrow/examples/README.md
@@ -24,5 +24,6 @@
 - [`dynamic_types.rs`](dynamic_types.rs): Dealing with mixed types dynamically at runtime
 - [`read_csv.rs`](read_csv.rs): Reading CSV files with explicit schema, pretty printing Arrays
 - [`read_csv_infer_schema.rs`](read_csv_infer_schema.rs): Reading CSV files, pretty printing Arrays
+- [`zero_copy_ipc`](zero_copy_ipc): Zero copy read of Arrow IPC file using `mmap`
 - [`tensor_builder.rs`](tensor_builder.rs): Using tensor builder
 - [`version.rs`](version.rs): Print the arrow version and exit
diff --git a/arrow/examples/zero_copy_ipc.rs b/arrow/examples/zero_copy_ipc.rs
new file mode 100644
index 000000000..15fc477c5
--- /dev/null
+++ b/arrow/examples/zero_copy_ipc.rs
@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Example of reading arrow IPC files and streams using "zero copy" API
+//!
+//! Zero copy in this case means the Arrow arrays refer directly to a user
+//! provided buffer or memory region.
+
+use arrow::array::{record_batch, RecordBatch};
+use arrow::error::Result;
+use arrow_buffer::Buffer;
+use arrow_cast::pretty::pretty_format_batches;
+use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::reader::{read_footer_length, FileDecoder};
+use arrow_ipc::writer::FileWriter;
+use arrow_ipc::{root_as_footer, Block};
+use std::path::PathBuf;
+use std::sync::Arc;
+
+/// This example shows how to read data from an Arrow IPC file without copying
+/// using `mmap` and the [`FileDecoder`] API
+fn main() {
+    // Create a temporary file with 3 record batches
+    let ipc_path = ipc_file();
+    // Open the file and memory map it using the mmap2 crate:
+    let ipc_file = std::fs::File::open(&ipc_path.path).expect("failed to open file");
+    let mmap = unsafe { memmap2::Mmap::map(&ipc_file).expect("failed to mmap file") };
+
+    // Convert the mmap region to an Arrow `Buffer` to back the arrow arrays. We
+    // do this by first creating a `bytes::Bytes` (which is zero copy) and then
+    // creating a Buffer from the `Bytes` (which is also zero copy)
+    let bytes = bytes::Bytes::from_owner(mmap);
+    let buffer = Buffer::from(bytes);
+
+    // Now, use the FileDecoder API (wrapped by `IPCBufferDecoder` for
+    // convenience) to crate Arrays re-using the data in the underlying buffer
+    let decoder = IPCBufferDecoder::new(buffer);
+    assert_eq!(decoder.num_batches(), 3);
+
+    // Create the Arrays and print them
+    for i in 0..decoder.num_batches() {
+        let batch = decoder.get_batch(i).unwrap().expect("failed to read batch");
+        assert_eq!(3, batch.num_rows());
+        println!("Batch {i}\n{}", pretty_format_batches(&[batch]).unwrap());
+    }
+}
+
+/// Return 3 [`RecordBatch`]es with a single column of type `Int32`
+fn example_data() -> Vec<RecordBatch> {
+    vec![
+        record_batch!(("my_column", Int32, [1, 2, 3])).unwrap(),
+        record_batch!(("my_column", Int32, [4, 5, 6])).unwrap(),
+        record_batch!(("my_column", Int32, [7, 8, 9])).unwrap(),
+    ]
+}
+
+/// Return a temporary file that contains an IPC file with 3 [`RecordBatch`]es
+fn ipc_file() -> TempFile {
+    let path = PathBuf::from("example.arrow");
+    let file = std::fs::File::create(&path).unwrap();
+    let data = example_data();
+    let mut writer = FileWriter::try_new(file, &data[0].schema()).unwrap();
+    for batch in &data {
+        writer.write(batch).unwrap();
+    }
+    writer.finish().unwrap();
+    TempFile { path }
+}
+
+/// Incrementally decodes [`RecordBatch`]es from an IPC file stored in a Arrow
+/// [`Buffer`] using the [`FileDecoder`] API.
+///
+/// This is a wrapper around the example in the `FileDecoder` which handles the
+/// low level interaction with the Arrow IPC format.
+struct IPCBufferDecoder {
+    /// Memory (or memory mapped) Buffer with the data
+    buffer: Buffer,
+    /// Decoder that reads Arrays that refers to the underlying buffers
+    decoder: FileDecoder,
+    /// Location of the batches within the buffer
+    batches: Vec<Block>,
+}
+
+impl IPCBufferDecoder {
+    fn new(buffer: Buffer) -> Self {
+        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());
+
+        let mut decoder = FileDecoder::new(Arc::new(schema), footer.version());
+
+        // Read dictionaries
+        for block in footer.dictionaries().iter().flatten() {
+            let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+            let data = buffer.slice_with_length(block.offset() as _, block_len);
+            decoder.read_dictionary(block, &data).unwrap();
+        }
+
+        // convert to Vec from the flatbuffers Vector to avoid having a direct dependency on flatbuffers
+        let batches = footer
+            .recordBatches()
+            .map(|b| b.iter().copied().collect())
+            .unwrap_or_default();
+
+        Self {
+            buffer,
+            decoder,
+            batches,
+        }
+    }
+
+    /// Return the number of [`RecordBatch`]es in this buffer
+    fn num_batches(&self) -> usize {
+        self.batches.len()
+    }
+
+    /// Return the [`RecordBatch`] at message index `i`.
+    ///
+    /// This may return `None` if the IPC message was None
+    fn get_batch(&self, i: usize) -> Result<Option<RecordBatch>> {
+        let block = &self.batches[i];
+        let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+        let data = self
+            .buffer
+            .slice_with_length(block.offset() as _, block_len);
+        self.decoder.read_record_batch(block, &data)
+    }
+}
+
+/// This structure deletes the file when it is dropped
+struct TempFile {
+    path: PathBuf,
+}
+
+impl Drop for TempFile {
+    fn drop(&mut self) {
+        if let Err(e) = std::fs::remove_file(&self.path) {
+            println!("Error deleting '{:?}': {:?}", self.path, e);
+        }
+    }
+}
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index fa5412c45..2c2e27ea4 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -1058,7 +1058,7 @@ impl TryFrom<Blob> for ObjectMeta {
         Ok(Self {
             location: Path::parse(value.name)?,
             last_modified: value.properties.last_modified,
-            size: value.properties.content_length as usize,
+            size: value.properties.content_length,
             e_tag: value.properties.e_tag,
             version: None, // For consistency with S3 and GCP which don't include this
         })
diff --git a/object_store/src/chunked.rs b/object_store/src/chunked.rs
index 4998e9f2a..2bb30b905 100644
--- a/object_store/src/chunked.rs
+++ b/object_store/src/chunked.rs
@@ -44,7 +44,7 @@ use crate::{PutPayload, Result};
 #[derive(Debug)]
 pub struct ChunkedStore {
     inner: Arc<dyn ObjectStore>,
-    chunk_size: usize,
+    chunk_size: usize, // chunks are in memory, so we use usize not u64
 }
 
 impl ChunkedStore {
@@ -138,7 +138,7 @@ impl ObjectStore for ChunkedStore {
         })
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         self.inner.get_range(location, range).await
     }
 
@@ -203,8 +203,8 @@ mod tests {
 
             let mut remaining = 1001;
             while let Some(next) = s.next().await {
-                let size = next.unwrap().len();
-                let expected = remaining.min(chunk_size);
+                let size = next.unwrap().len() as u64;
+                let expected = remaining.min(chunk_size as u64);
                 assert_eq!(size, expected);
                 remaining -= expected;
             }
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 57aca8956..f252dd9c2 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -67,9 +67,9 @@ impl<T: GetClient> GetClientExt for T {
 
 struct ContentRange {
     /// The range of the object returned
-    range: Range<usize>,
+    range: Range<u64>,
     /// The total size of the object being requested
-    size: usize,
+    size: u64,
 }
 
 impl ContentRange {
@@ -84,7 +84,7 @@ impl ContentRange {
         let (start_s, end_s) = range.split_once('-')?;
 
         let start = start_s.parse().ok()?;
-        let end: usize = end_s.parse().ok()?;
+        let end: u64 = end_s.parse().ok()?;
 
         Some(Self {
             size,
@@ -140,8 +140,8 @@ enum GetResultError {
 
     #[error("Requested {expected:?}, got {actual:?}")]
     UnexpectedRange {
-        expected: Range<usize>,
-        actual: Range<usize>,
+        expected: Range<u64>,
+        actual: Range<u64>,
     },
 }
 
diff --git a/object_store/src/client/s3.rs b/object_store/src/client/s3.rs
index 7fe956b23..a2221fbbc 100644
--- a/object_store/src/client/s3.rs
+++ b/object_store/src/client/s3.rs
@@ -66,7 +66,7 @@ pub struct ListPrefix {
 #[serde(rename_all = "PascalCase")]
 pub struct ListContents {
     pub key: String,
-    pub size: usize,
+    pub size: u64,
     pub last_modified: DateTime<Utc>,
     #[serde(rename = "ETag")]
     pub e_tag: Option<String>,
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index 41e6464c1..9983fdff5 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -420,7 +420,7 @@ impl MultiStatusResponse {
         })?)
     }
 
-    fn size(&self) -> Result<usize> {
+    fn size(&self) -> Result<u64> {
         let size = self
             .prop_stat
             .prop
@@ -462,7 +462,7 @@ pub(crate) struct Prop {
     last_modified: DateTime<Utc>,
 
     #[serde(rename = "getcontentlength")]
-    content_length: Option<usize>,
+    content_length: Option<u64>,
 
     #[serde(rename = "resourcetype")]
     resource_type: ResourceType,
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
index 20e95fddc..25a929459 100644
--- a/object_store/src/integration.rs
+++ b/object_store/src/integration.rs
@@ -112,7 +112,7 @@ pub async fn put_get_delete_list(storage: &DynObjectStore) {
     let range_result = storage.get_range(&location, range.clone()).await;
 
     let bytes = range_result.unwrap();
-    assert_eq!(bytes, data.slice(range.clone()));
+    assert_eq!(bytes, data.slice(range.start as usize..range.end as usize));
 
     let opts = GetOptions {
         range: Some(GetRange::Bounded(2..5)),
@@ -190,11 +190,11 @@ pub async fn put_get_delete_list(storage: &DynObjectStore) {
     let ranges = vec![0..1, 2..3, 0..5];
     let bytes = storage.get_ranges(&location, &ranges).await.unwrap();
     for (range, bytes) in ranges.iter().zip(bytes) {
-        assert_eq!(bytes, data.slice(range.clone()))
+        assert_eq!(bytes, data.slice(range.start as usize..range.end as usize));
     }
 
     let head = storage.head(&location).await.unwrap();
-    assert_eq!(head.size, data.len());
+    assert_eq!(head.size, data.len() as u64);
 
     storage.delete(&location).await.unwrap();
 
@@ -934,7 +934,7 @@ pub async fn list_with_delimiter(storage: &DynObjectStore) {
     let object = &result.objects[0];
 
     assert_eq!(object.location, expected_location);
-    assert_eq!(object.size, data.len());
+    assert_eq!(object.size, data.len() as u64);
 
     // ==================== check: prefix-list `mydb/wb/000/000/001` (partial filename doesn't match) ====================
     let prefix = Path::from("mydb/wb/000/000/001");
@@ -1085,7 +1085,7 @@ pub async fn multipart(storage: &dyn ObjectStore, multipart: &dyn MultipartStore
         .unwrap();
 
     let meta = storage.head(&path).await.unwrap();
-    assert_eq!(meta.size, chunk_size * 2);
+    assert_eq!(meta.size, chunk_size as u64 * 2);
 
     // Empty case
     let path = Path::from("test_empty_multipart");
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 53eda5a82..cffcbbdd4 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -234,7 +234,7 @@
 //!
 //! // Buffer the entire object in memory
 //! let object: Bytes = result.bytes().await.unwrap();
-//! assert_eq!(object.len(), meta.size);
+//! assert_eq!(object.len() as u64, meta.size);
 //!
 //! // Alternatively stream the bytes from object storage
 //! let stream = object_store.get(&path).await.unwrap().into_stream();
@@ -630,7 +630,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
     /// in the given byte range.
     ///
     /// See [`GetRange::Bounded`] for more details on how `range` gets interpreted
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let options = GetOptions {
             range: Some(range.into()),
             ..Default::default()
@@ -640,7 +640,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
 
     /// Return the bytes that are stored at the specified location
     /// in the given byte ranges
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         coalesce_ranges(
             ranges,
             |range| self.get_range(location, range),
@@ -820,14 +820,14 @@ macro_rules! as_ref_impl {
                 self.as_ref().get_opts(location, options).await
             }
 
-            async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+            async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
                 self.as_ref().get_range(location, range).await
             }
 
             async fn get_ranges(
                 &self,
                 location: &Path,
-                ranges: &[Range<usize>],
+                ranges: &[Range<u64>],
             ) -> Result<Vec<Bytes>> {
                 self.as_ref().get_ranges(location, ranges).await
             }
@@ -903,8 +903,10 @@ pub struct ObjectMeta {
     pub location: Path,
     /// The last modified time
     pub last_modified: DateTime<Utc>,
-    /// The size in bytes of the object
-    pub size: usize,
+    /// The size in bytes of the object.
+    ///
+    /// Note this is not `usize` as `object_store` supports 32-bit architectures such as WASM
+    pub size: u64,
     /// The unique identifier for the object
     ///
     /// <https://datatracker.ietf.org/doc/html/rfc9110#name-etag>
@@ -1019,7 +1021,9 @@ pub struct GetResult {
     /// The [`ObjectMeta`] for this object
     pub meta: ObjectMeta,
     /// The range of bytes returned by this request
-    pub range: Range<usize>,
+    ///
+    /// Note this is not `usize` as `object_store` supports 32-bit architectures such as WASM
+    pub range: Range<u64>,
     /// Additional object attributes
     pub attributes: Attributes,
 }
@@ -1060,7 +1064,11 @@ impl GetResult {
                             path: path.clone(),
                         })?;
 
-                    let mut buffer = Vec::with_capacity(len);
+                    let mut buffer = if let Ok(len) = len.try_into() {
+                        Vec::with_capacity(len)
+                    } else {
+                        Vec::new()
+                    };
                     file.take(len as _)
                         .read_to_end(&mut buffer)
                         .map_err(|source| local::Error::UnableToReadBytes { source, path })?;
diff --git a/object_store/src/limit.rs b/object_store/src/limit.rs
index 77f72a0e1..330a0da07 100644
--- a/object_store/src/limit.rs
+++ b/object_store/src/limit.rs
@@ -117,12 +117,12 @@ impl<T: ObjectStore> ObjectStore for LimitStore<T> {
         Ok(permit_get_result(r, permit))
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let _permit = self.semaphore.acquire().await.unwrap();
         self.inner.get_range(location, range).await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let _permit = self.semaphore.acquire().await.unwrap();
         self.inner.get_ranges(location, ranges).await
     }
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 364026459..65e87f982 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -21,7 +21,7 @@ use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
 use std::ops::Range;
 use std::sync::Arc;
 use std::time::SystemTime;
-use std::{collections::BTreeSet, convert::TryFrom, io};
+use std::{collections::BTreeSet, io};
 use std::{collections::VecDeque, path::PathBuf};
 
 use async_trait::async_trait;
@@ -44,12 +44,6 @@ use crate::{
 /// A specialized `Error` for filesystem object store-related errors
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
-    #[error("File size for {} did not fit in a usize: {}", path, source)]
-    FileSizeOverflowedUsize {
-        source: std::num::TryFromIntError,
-        path: String,
-    },
-
     #[error("Unable to walk dir: {}", source)]
     UnableToWalkDir { source: walkdir::Error },
 
@@ -83,8 +77,8 @@ pub(crate) enum Error {
     #[error("Out of range of file {}, expected: {}, actual: {}", path.display(), expected, actual)]
     OutOfRange {
         path: PathBuf,
-        expected: usize,
-        actual: usize,
+        expected: u64,
+        actual: u64,
     },
 
     #[error("Requested range was invalid")]
@@ -410,7 +404,7 @@ impl ObjectStore for LocalFileSystem {
         let path = self.path_to_filesystem(&location)?;
         maybe_spawn_blocking(move || {
             let (file, metadata) = open_file(&path)?;
-            let meta = convert_metadata(metadata, location)?;
+            let meta = convert_metadata(metadata, location);
             options.check_preconditions(&meta)?;
 
             let range = match options.range {
@@ -430,7 +424,7 @@ impl ObjectStore for LocalFileSystem {
         .await
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let path = self.path_to_filesystem(location)?;
         maybe_spawn_blocking(move || {
             let (mut file, _) = open_file(&path)?;
@@ -439,7 +433,7 @@ impl ObjectStore for LocalFileSystem {
         .await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let path = self.path_to_filesystem(location)?;
         let ranges = ranges.to_vec();
         maybe_spawn_blocking(move || {
@@ -825,7 +819,7 @@ impl Drop for LocalUpload {
 pub(crate) fn chunked_stream(
     mut file: File,
     path: PathBuf,
-    range: Range<usize>,
+    range: Range<u64>,
     chunk_size: usize,
 ) -> BoxStream<'static, Result<Bytes, super::Error>> {
     futures::stream::once(async move {
@@ -847,17 +841,23 @@ pub(crate) fn chunked_stream(
                         return Ok(None);
                     }
 
-                    let to_read = remaining.min(chunk_size);
-                    let mut buffer = Vec::with_capacity(to_read);
+                    let to_read = remaining.min(chunk_size as u64);
+                    let cap = usize::try_from(to_read).map_err(|_e| Error::InvalidRange {
+                        source: InvalidGetRange::TooLarge {
+                            requested: to_read,
+                            max: usize::MAX as u64,
+                        },
+                    })?;
+                    let mut buffer = Vec::with_capacity(cap);
                     let read = (&mut file)
-                        .take(to_read as u64)
+                        .take(to_read)
                         .read_to_end(&mut buffer)
                         .map_err(|e| Error::UnableToReadBytes {
                             source: e,
                             path: path.clone(),
                         })?;
 
-                    Ok(Some((buffer.into(), (file, path, remaining - read))))
+                    Ok(Some((buffer.into(), (file, path, remaining - read as u64))))
                 })
             },
         );
@@ -867,22 +867,18 @@ pub(crate) fn chunked_stream(
     .boxed()
 }
 
-pub(crate) fn read_range(file: &mut File, path: &PathBuf, range: Range<usize>) -> Result<Bytes> {
+pub(crate) fn read_range(file: &mut File, path: &PathBuf, range: Range<u64>) -> Result<Bytes> {
     let to_read = range.end - range.start;
-    file.seek(SeekFrom::Start(range.start as u64))
-        .map_err(|source| {
-            let path = path.into();
-            Error::Seek { source, path }
-        })?;
+    file.seek(SeekFrom::Start(range.start)).map_err(|source| {
+        let path = path.into();
+        Error::Seek { source, path }
+    })?;
 
-    let mut buf = Vec::with_capacity(to_read);
-    let read = file
-        .take(to_read as u64)
-        .read_to_end(&mut buf)
-        .map_err(|source| {
-            let path = path.into();
-            Error::UnableToReadBytes { source, path }
-        })?;
+    let mut buf = Vec::with_capacity(to_read as usize);
+    let read = file.take(to_read).read_to_end(&mut buf).map_err(|source| {
+        let path = path.into();
+        Error::UnableToReadBytes { source, path }
+    })? as u64;
 
     if read != to_read {
         let error = Error::OutOfRange {
@@ -922,7 +918,7 @@ fn open_file(path: &PathBuf) -> Result<(File, Metadata)> {
 
 fn convert_entry(entry: DirEntry, location: Path) -> Result<Option<ObjectMeta>> {
     match entry.metadata() {
-        Ok(metadata) => convert_metadata(metadata, location).map(Some),
+        Ok(metadata) => Ok(Some(convert_metadata(metadata, location))),
         Err(e) => {
             if let Some(io_err) = e.io_error() {
                 if io_err.kind() == ErrorKind::NotFound {
@@ -960,20 +956,16 @@ fn get_etag(metadata: &Metadata) -> String {
     format!("{inode:x}-{mtime:x}-{size:x}")
 }
 
-fn convert_metadata(metadata: Metadata, location: Path) -> Result<ObjectMeta> {
+fn convert_metadata(metadata: Metadata, location: Path) -> ObjectMeta {
     let last_modified = last_modified(&metadata);
-    let size = usize::try_from(metadata.len()).map_err(|source| {
-        let path = location.as_ref().into();
-        Error::FileSizeOverflowedUsize { source, path }
-    })?;
 
-    Ok(ObjectMeta {
+    ObjectMeta {
         location,
         last_modified,
-        size,
+        size: metadata.len(),
         e_tag: Some(get_etag(&metadata)),
         version: None,
-    })
+    }
 }
 
 #[cfg(unix)]
diff --git a/object_store/src/memory.rs b/object_store/src/memory.rs
index 6402f9243..26beff1e1 100644
--- a/object_store/src/memory.rs
+++ b/object_store/src/memory.rs
@@ -241,7 +241,7 @@ impl ObjectStore for InMemory {
         let meta = ObjectMeta {
             location: location.clone(),
             last_modified: entry.last_modified,
-            size: entry.data.len(),
+            size: entry.data.len() as u64,
             e_tag: Some(e_tag),
             version: None,
         };
@@ -250,11 +250,14 @@ impl ObjectStore for InMemory {
         let (range, data) = match options.range {
             Some(range) => {
                 let r = range
-                    .as_range(entry.data.len())
+                    .as_range(entry.data.len() as u64)
                     .map_err(|source| Error::Range { source })?;
-                (r.clone(), entry.data.slice(r))
+                (
+                    r.clone(),
+                    entry.data.slice(r.start as usize..r.end as usize),
+                )
             }
-            None => (0..entry.data.len(), entry.data),
+            None => (0..entry.data.len() as u64, entry.data),
         };
         let stream = futures::stream::once(futures::future::ready(Ok(data)));
 
@@ -266,16 +269,27 @@ impl ObjectStore for InMemory {
         })
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let entry = self.entry(location).await?;
         ranges
             .iter()
             .map(|range| {
                 let r = GetRange::Bounded(range.clone())
-                    .as_range(entry.data.len())
+                    .as_range(entry.data.len() as u64)
                     .map_err(|source| Error::Range { source })?;
-
-                Ok(entry.data.slice(r))
+                let r_end = usize::try_from(r.end).map_err(|_e| Error::Range {
+                    source: InvalidGetRange::TooLarge {
+                        requested: r.end,
+                        max: usize::MAX as u64,
+                    },
+                })?;
+                let r_start = usize::try_from(r.start).map_err(|_e| Error::Range {
+                    source: InvalidGetRange::TooLarge {
+                        requested: r.start,
+                        max: usize::MAX as u64,
+                    },
+                })?;
+                Ok(entry.data.slice(r_start..r_end))
             })
             .collect()
     }
@@ -286,7 +300,7 @@ impl ObjectStore for InMemory {
         Ok(ObjectMeta {
             location: location.clone(),
             last_modified: entry.last_modified,
-            size: entry.data.len(),
+            size: entry.data.len() as u64,
             e_tag: Some(entry.e_tag.to_string()),
             version: None,
         })
@@ -316,7 +330,7 @@ impl ObjectStore for InMemory {
                 Ok(ObjectMeta {
                     location: key.clone(),
                     last_modified: value.last_modified,
-                    size: value.data.len(),
+                    size: value.data.len() as u64,
                     e_tag: Some(value.e_tag.to_string()),
                     version: None,
                 })
@@ -361,7 +375,7 @@ impl ObjectStore for InMemory {
                 let object = ObjectMeta {
                     location: k.clone(),
                     last_modified: v.last_modified,
-                    size: v.data.len(),
+                    size: v.data.len() as u64,
                     e_tag: Some(v.e_tag.to_string()),
                     version: None,
                 };
diff --git a/object_store/src/prefix.rs b/object_store/src/prefix.rs
index a0b67ca4b..ac9803ec0 100644
--- a/object_store/src/prefix.rs
+++ b/object_store/src/prefix.rs
@@ -132,7 +132,7 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
         self.inner.get(&full_path).await
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let full_path = self.full_path(location);
         self.inner.get_range(&full_path, range).await
     }
@@ -142,7 +142,7 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
         self.inner.get_opts(&full_path, options).await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let full_path = self.full_path(location);
         self.inner.get_ranges(&full_path, ranges).await
     }
diff --git a/object_store/src/throttle.rs b/object_store/src/throttle.rs
index 29cd32705..6586ba946 100644
--- a/object_store/src/throttle.rs
+++ b/object_store/src/throttle.rs
@@ -203,7 +203,7 @@ impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
         Ok(throttle_get(result, wait_get_per_byte))
     }
 
-    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> Result<Bytes> {
         let config = self.config();
 
         let sleep_duration =
@@ -214,10 +214,10 @@ impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
         self.inner.get_range(location, range).await
     }
 
-    async fn get_ranges(&self, location: &Path, ranges: &[Range<usize>]) -> Result<Vec<Bytes>> {
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
         let config = self.config();
 
-        let total_bytes: usize = ranges.iter().map(|range| range.end - range.start).sum();
+        let total_bytes: u64 = ranges.iter().map(|range| range.end - range.start).sum();
         let sleep_duration =
             config.wait_get_per_call + config.wait_get_per_byte * total_bytes as u32;
 
diff --git a/object_store/src/util.rs b/object_store/src/util.rs
index 6d638f3cb..17a7a8cad 100644
--- a/object_store/src/util.rs
+++ b/object_store/src/util.rs
@@ -49,7 +49,7 @@ pub(crate) fn hmac_sha256(secret: impl AsRef<[u8]>, bytes: impl AsRef<[u8]>) ->
 }
 
 /// Collect a stream into [`Bytes`] avoiding copying in the event of a single chunk
-pub async fn collect_bytes<S, E>(mut stream: S, size_hint: Option<usize>) -> Result<Bytes, E>
+pub async fn collect_bytes<S, E>(mut stream: S, size_hint: Option<u64>) -> Result<Bytes, E>
 where
     E: Send,
     S: Stream<Item = Result<Bytes, E>> + Send + Unpin,
@@ -60,9 +60,9 @@ where
     match stream.next().await.transpose()? {
         None => Ok(first),
         Some(second) => {
-            let size_hint = size_hint.unwrap_or_else(|| first.len() + second.len());
+            let size_hint = size_hint.unwrap_or_else(|| first.len() as u64 + second.len() as u64);
 
-            let mut buf = Vec::with_capacity(size_hint);
+            let mut buf = Vec::with_capacity(size_hint as usize);
             buf.extend_from_slice(&first);
             buf.extend_from_slice(&second);
             while let Some(maybe_bytes) = stream.next().await {
@@ -89,7 +89,7 @@ where
 
 /// Range requests with a gap less than or equal to this,
 /// will be coalesced into a single request by [`coalesce_ranges`]
-pub const OBJECT_STORE_COALESCE_DEFAULT: usize = 1024 * 1024;
+pub const OBJECT_STORE_COALESCE_DEFAULT: u64 = 1024 * 1024;
 
 /// Up to this number of range requests will be performed in parallel by [`coalesce_ranges`]
 pub(crate) const OBJECT_STORE_COALESCE_PARALLEL: usize = 10;
@@ -103,12 +103,12 @@ pub(crate) const OBJECT_STORE_COALESCE_PARALLEL: usize = 10;
 /// * Make multiple `fetch` requests in parallel (up to maximum of 10)
 ///
 pub async fn coalesce_ranges<F, E, Fut>(
-    ranges: &[Range<usize>],
+    ranges: &[Range<u64>],
     fetch: F,
-    coalesce: usize,
+    coalesce: u64,
 ) -> Result<Vec<Bytes>, E>
 where
-    F: Send + FnMut(Range<usize>) -> Fut,
+    F: Send + FnMut(Range<u64>) -> Fut,
     E: Send,
     Fut: std::future::Future<Output = Result<Bytes, E>> + Send,
 {
@@ -129,13 +129,14 @@ where
 
             let start = range.start - fetch_range.start;
             let end = range.end - fetch_range.start;
-            fetch_bytes.slice(start..end.min(fetch_bytes.len()))
+            let range = (start as usize)..(end as usize).min(fetch_bytes.len());
+            fetch_bytes.slice(range)
         })
         .collect())
 }
 
 /// Returns a sorted list of ranges that cover `ranges`
-fn merge_ranges(ranges: &[Range<usize>], coalesce: usize) -> Vec<Range<usize>> {
+fn merge_ranges(ranges: &[Range<u64>], coalesce: u64) -> Vec<Range<u64>> {
     if ranges.is_empty() {
         return vec![];
     }
@@ -196,38 +197,49 @@ pub enum GetRange {
     /// an error will be returned. Additionally, if the range ends after the end
     /// of the object, the entire remainder of the object will be returned.
     /// Otherwise, the exact requested range will be returned.
-    Bounded(Range<usize>),
+    ///
+    /// Note that range is u64 (i.e., not usize),
+    /// as `object_store` supports 32-bit architectures such as WASM
+    Bounded(Range<u64>),
     /// Request all bytes starting from a given byte offset
-    Offset(usize),
+    Offset(u64),
     /// Request up to the last n bytes
-    Suffix(usize),
+    Suffix(u64),
 }
 
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum InvalidGetRange {
     #[error("Wanted range starting at {requested}, but object was only {length} bytes long")]
-    StartTooLarge { requested: usize, length: usize },
+    StartTooLarge { requested: u64, length: u64 },
 
     #[error("Range started at {start} and ended at {end}")]
-    Inconsistent { start: usize, end: usize },
+    Inconsistent { start: u64, end: u64 },
+
+    #[error("Range {requested} is larger than system memory limit {max}")]
+    TooLarge { requested: u64, max: u64 },
 }
 
 impl GetRange {
     pub(crate) fn is_valid(&self) -> Result<(), InvalidGetRange> {
-        match self {
-            Self::Bounded(r) if r.end <= r.start => {
+        if let Self::Bounded(r) = self {
+            if r.end <= r.start {
                 return Err(InvalidGetRange::Inconsistent {
                     start: r.start,
                     end: r.end,
                 });
             }
-            _ => (),
-        };
+            if (r.end - r.start) > usize::MAX as u64 {
+                return Err(InvalidGetRange::TooLarge {
+                    requested: r.start,
+                    max: usize::MAX as u64,
+                });
+            }
+        }
         Ok(())
     }
 
     /// Convert to a [`Range`] if valid.
-    pub(crate) fn as_range(&self, len: usize) -> Result<Range<usize>, InvalidGetRange> {
+    pub(crate) fn as_range(&self, len: u64) -> Result<Range<u64>, InvalidGetRange> {
         self.is_valid()?;
         match self {
             Self::Bounded(r) => {
@@ -267,7 +279,7 @@ impl Display for GetRange {
     }
 }
 
-impl<T: RangeBounds<usize>> From<T> for GetRange {
+impl<T: RangeBounds<u64>> From<T> for GetRange {
     fn from(value: T) -> Self {
         use std::ops::Bound::*;
         let first = match value.start_bound() {
@@ -323,7 +335,7 @@ mod tests {
     /// Calls coalesce_ranges and validates the returned data is correct
     ///
     /// Returns the fetched ranges
-    async fn do_fetch(ranges: Vec<Range<usize>>, coalesce: usize) -> Vec<Range<usize>> {
+    async fn do_fetch(ranges: Vec<Range<u64>>, coalesce: u64) -> Vec<Range<u64>> {
         let max = ranges.iter().map(|x| x.end).max().unwrap_or(0);
         let src: Vec<_> = (0..max).map(|x| x as u8).collect();
 
@@ -332,7 +344,9 @@ mod tests {
             &ranges,
             |range| {
                 fetches.push(range.clone());
-                futures::future::ready(Ok(Bytes::from(src[range].to_vec())))
+                let start = usize::try_from(range.start).unwrap();
+                let end = usize::try_from(range.end).unwrap();
+                futures::future::ready(Ok(Bytes::from(src[start..end].to_vec())))
             },
             coalesce,
         )
@@ -341,7 +355,10 @@ mod tests {
 
         assert_eq!(ranges.len(), coalesced.len());
         for (range, bytes) in ranges.iter().zip(coalesced) {
-            assert_eq!(bytes.as_ref(), &src[range.clone()]);
+            assert_eq!(
+                bytes.as_ref(),
+                &src[usize::try_from(range.start).unwrap()..usize::try_from(range.end).unwrap()]
+            );
         }
         fetches
     }
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index c14c0e1d3..54992d864 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -69,7 +69,7 @@ paste = { version = "1.0" }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
 sysinfo = { version = "0.33.0", optional = true, default-features = false, features = ["system"] }
 crc32fast = { version = "1.4.2", optional = true, default-features = false }
-simdutf8 = { version = "0.1.5"}
+simdutf8 = { version = "0.1.5", optional = true, default-features = false }
 
 [dev-dependencies]
 base64 = { version = "0.22", default-features = false, features = ["std"] }
@@ -98,7 +98,7 @@ zstd-sys = { version = ">=2.0.0, <2.0.14", default-features = false }
 all-features = true
 
 [features]
-default = ["arrow", "snap", "brotli", "flate2", "lz4", "zstd", "base64"]
+default = ["arrow", "snap", "brotli", "flate2", "lz4", "zstd", "base64", "simdutf8"]
 # Enable lz4
 lz4 = ["lz4_flex"]
 # Enable arrow reader/writer APIs
@@ -121,6 +121,8 @@ zstd = ["dep:zstd", "zstd-sys"]
 sysinfo = ["dep:sysinfo"]
 # Verify 32-bit CRC checksum when decoding parquet pages
 crc = ["dep:crc32fast"]
+# Enable SIMD UTF-8 validation
+simdutf8 = ["dep:simdutf8"]
 
 
 [[example]]
diff --git a/parquet/README.md b/parquet/README.md
index 9ff1d921d..1224e52f3 100644
--- a/parquet/README.md
+++ b/parquet/README.md
@@ -51,17 +51,21 @@ major releases may contain breaking API changes.
 
 The `parquet` crate provides the following features which may be enabled in your `Cargo.toml`:
 
-- `arrow` (default) - support for reading / writing [`arrow`](https://crates.io/crates/arrow) arrays to / from parquet
-- `async` - support `async` APIs for reading parquet
-- `json` - support for reading / writing `json` data to / from parquet
-- `brotli` (default) - support for parquet using `brotli` compression
-- `flate2` (default) - support for parquet using `gzip` compression
-- `lz4` (default) - support for parquet using `lz4` compression
-- `zstd` (default) - support for parquet using `zstd` compression
-- `snap` (default) - support for parquet using `snappy` compression
+- `arrow` (default) - support for reading / writing [`arrow`] arrays to / from Parquet
+- `async` - support `async` APIs for reading Parquet
+- `json` - support for reading / writing `json` data to / from Parquet
+- `brotli` (default) - support for Parquet using `brotli` compression
+- `flate2` (default) - support for Parquet using `gzip` compression
+- `lz4` (default) - support for Parquet using `lz4` compression
+- `zstd` (default) - support for Parquet using `zstd` compression
+- `snap` (default) - support for Parquet using `snappy` compression
 - `cli` - parquet [CLI tools](https://github.com/apache/arrow-rs/tree/main/parquet/src/bin)
 - `crc` - enables functionality to automatically verify checksums of each page (if present) when decoding
 - `experimental` - Experimental APIs which may change, even between minor releases
+- `simdutf8` (default) - Use the [`simdutf8`] crate for SIMD-accelerated UTF-8 validation
+
+[`arrow`]: https://crates.io/crates/arrow
+[`simdutf8`]: https://crates.io/crates/simdutf8
 
 ## Parquet Feature Status
 
diff --git a/parquet/src/arrow/array_reader/byte_view_array.rs b/parquet/src/arrow/array_reader/byte_view_array.rs
index 00627ad61..8df659060 100644
--- a/parquet/src/arrow/array_reader/byte_view_array.rs
+++ b/parquet/src/arrow/array_reader/byte_view_array.rs
@@ -27,6 +27,7 @@ use crate::data_type::Int32Type;
 use crate::encodings::decoding::{Decoder, DeltaBitPackDecoder};
 use crate::errors::{ParquetError, Result};
 use crate::schema::types::ColumnDescPtr;
+use crate::util::utf8::check_valid_utf8;
 use arrow_array::{builder::make_view, ArrayRef};
 use arrow_buffer::Buffer;
 use arrow_data::ByteView;
@@ -681,17 +682,6 @@ impl ByteViewArrayDecoderDelta {
     }
 }
 
-/// Check that `val` is a valid UTF-8 sequence
-pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
-    match simdutf8::basic::from_utf8(val) {
-        Ok(_) => Ok(()),
-        Err(_) => {
-            let e = simdutf8::compat::from_utf8(val).unwrap_err();
-            Err(general_err!("encountered non UTF-8 data: {}", e))
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use arrow_array::StringViewArray;
diff --git a/parquet/src/arrow/array_reader/primitive_array.rs b/parquet/src/arrow/array_reader/primitive_array.rs
index a952e00e1..709d0f8bb 100644
--- a/parquet/src/arrow/array_reader/primitive_array.rs
+++ b/parquet/src/arrow/array_reader/primitive_array.rs
@@ -270,6 +270,59 @@ where
 
                 Arc::new(array) as ArrayRef
             }
+            ArrowType::Dictionary(_, value_type) => match value_type.as_ref() {
+                ArrowType::Decimal128(p, s) => {
+                    let array = match array.data_type() {
+                        ArrowType::Int32 => array
+                            .as_any()
+                            .downcast_ref::<Int32Array>()
+                            .unwrap()
+                            .unary(|i| i as i128)
+                            as Decimal128Array,
+                        ArrowType::Int64 => array
+                            .as_any()
+                            .downcast_ref::<Int64Array>()
+                            .unwrap()
+                            .unary(|i| i as i128)
+                            as Decimal128Array,
+                        _ => {
+                            return Err(arrow_err!(
+                                "Cannot convert {:?} to decimal dictionary",
+                                array.data_type()
+                            ));
+                        }
+                    }
+                    .with_precision_and_scale(*p, *s)?;
+
+                    arrow_cast::cast(&array, target_type)?
+                }
+                ArrowType::Decimal256(p, s) => {
+                    let array = match array.data_type() {
+                        ArrowType::Int32 => array
+                            .as_any()
+                            .downcast_ref::<Int32Array>()
+                            .unwrap()
+                            .unary(i256::from)
+                            as Decimal256Array,
+                        ArrowType::Int64 => array
+                            .as_any()
+                            .downcast_ref::<Int64Array>()
+                            .unwrap()
+                            .unary(i256::from)
+                            as Decimal256Array,
+                        _ => {
+                            return Err(arrow_err!(
+                                "Cannot convert {:?} to decimal dictionary",
+                                array.data_type()
+                            ));
+                        }
+                    }
+                    .with_precision_and_scale(*p, *s)?;
+
+                    arrow_cast::cast(&array, target_type)?
+                }
+                _ => arrow_cast::cast(&array, target_type)?,
+            },
             _ => arrow_cast::cast(&array, target_type)?,
         };
 
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 41f15569f..f1081f148 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -843,6 +843,27 @@ fn write_leaf(writer: &mut ColumnWriter<'_>, levels: &ArrayLevels) -> Result<usi
                         .unary::<_, Int32Type>(|v| v.as_i128() as i32);
                     write_primitive(typed, array.values(), levels)
                 }
+                ArrowDataType::Dictionary(_, value_type) => match value_type.as_ref() {
+                    ArrowDataType::Decimal128(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal128Type>()
+                            .unary::<_, Int32Type>(|v| v as i32);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    ArrowDataType::Decimal256(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal256Type>()
+                            .unary::<_, Int32Type>(|v| v.as_i128() as i32);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    _ => {
+                        let array = arrow_cast::cast(column, &ArrowDataType::Int32)?;
+                        let array = array.as_primitive::<Int32Type>();
+                        write_primitive(typed, array.values(), levels)
+                    }
+                },
                 _ => {
                     let array = arrow_cast::cast(column, &ArrowDataType::Int32)?;
                     let array = array.as_primitive::<Int32Type>();
@@ -891,6 +912,27 @@ fn write_leaf(writer: &mut ColumnWriter<'_>, levels: &ArrayLevels) -> Result<usi
                         .unary::<_, Int64Type>(|v| v.as_i128() as i64);
                     write_primitive(typed, array.values(), levels)
                 }
+                ArrowDataType::Dictionary(_, value_type) => match value_type.as_ref() {
+                    ArrowDataType::Decimal128(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal128Type>()
+                            .unary::<_, Int64Type>(|v| v as i64);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    ArrowDataType::Decimal256(_, _) => {
+                        let array = arrow_cast::cast(column, value_type)?;
+                        let array = array
+                            .as_primitive::<Decimal256Type>()
+                            .unary::<_, Int64Type>(|v| v.as_i128() as i64);
+                        write_primitive(typed, array.values(), levels)
+                    }
+                    _ => {
+                        let array = arrow_cast::cast(column, &ArrowDataType::Int64)?;
+                        let array = array.as_primitive::<Int64Type>();
+                        write_primitive(typed, array.values(), levels)
+                    }
+                },
                 _ => {
                     let array = arrow_cast::cast(column, &ArrowDataType::Int64)?;
                     let array = array.as_primitive::<Int64Type>();
@@ -1093,7 +1135,7 @@ mod tests {
     use arrow::util::data_gen::create_random_array;
     use arrow::util::pretty::pretty_format_batches;
     use arrow::{array::*, buffer::Buffer};
-    use arrow_buffer::{IntervalDayTime, IntervalMonthDayNano, NullBuffer};
+    use arrow_buffer::{i256, IntervalDayTime, IntervalMonthDayNano, NullBuffer};
     use arrow_schema::Fields;
     use half::f16;
 
@@ -2670,6 +2712,52 @@ mod tests {
         one_column_roundtrip_with_schema(Arc::new(d), schema);
     }
 
+    #[test]
+    fn arrow_writer_decimal128_dictionary() {
+        let integers = vec![12345, 56789, 34567];
+
+        let keys = UInt8Array::from(vec![Some(0), None, Some(1), Some(2), Some(1)]);
+
+        let values = Decimal128Array::from(integers.clone())
+            .with_precision_and_scale(5, 2)
+            .unwrap();
+
+        let array = DictionaryArray::new(keys, Arc::new(values));
+        one_column_roundtrip(Arc::new(array.clone()), true);
+
+        let values = Decimal128Array::from(integers)
+            .with_precision_and_scale(12, 2)
+            .unwrap();
+
+        let array = array.with_values(Arc::new(values));
+        one_column_roundtrip(Arc::new(array), true);
+    }
+
+    #[test]
+    fn arrow_writer_decimal256_dictionary() {
+        let integers = vec![
+            i256::from_i128(12345),
+            i256::from_i128(56789),
+            i256::from_i128(34567),
+        ];
+
+        let keys = UInt8Array::from(vec![Some(0), None, Some(1), Some(2), Some(1)]);
+
+        let values = Decimal256Array::from(integers.clone())
+            .with_precision_and_scale(5, 2)
+            .unwrap();
+
+        let array = DictionaryArray::new(keys, Arc::new(values));
+        one_column_roundtrip(Arc::new(array.clone()), true);
+
+        let values = Decimal256Array::from(integers)
+            .with_precision_and_scale(12, 2)
+            .unwrap();
+
+        let array = array.with_values(Arc::new(values));
+        one_column_roundtrip(Arc::new(array), true);
+    }
+
     #[test]
     fn arrow_writer_string_dictionary_unsigned_index() {
         // define schema
@@ -3409,4 +3497,34 @@ mod tests {
             "Arrow: Incompatible type. Field 'temperature' has type Float64, array has type Int32"
         );
     }
+
+    #[test]
+    // https://github.com/apache/arrow-rs/issues/6988
+    fn test_roundtrip_empty_schema() {
+        // create empty record batch with empty schema
+        let empty_batch = RecordBatch::try_new_with_options(
+            Arc::new(Schema::empty()),
+            vec![],
+            &RecordBatchOptions::default().with_row_count(Some(0)),
+        )
+        .unwrap();
+
+        // write to parquet
+        let mut parquet_bytes: Vec<u8> = Vec::new();
+        let mut writer =
+            ArrowWriter::try_new(&mut parquet_bytes, empty_batch.schema(), None).unwrap();
+        writer.write(&empty_batch).unwrap();
+        writer.close().unwrap();
+
+        // read from parquet
+        let bytes = Bytes::from(parquet_bytes);
+        let reader = ParquetRecordBatchReaderBuilder::try_new(bytes).unwrap();
+        assert_eq!(reader.schema(), &empty_batch.schema());
+        let batches: Vec<_> = reader
+            .build()
+            .unwrap()
+            .collect::<ArrowResult<Vec<_>>>()
+            .unwrap();
+        assert_eq!(batches.len(), 0);
+    }
 }
diff --git a/parquet/src/arrow/buffer/offset_buffer.rs b/parquet/src/arrow/buffer/offset_buffer.rs
index 8dfb85961..5051dce12 100644
--- a/parquet/src/arrow/buffer/offset_buffer.rs
+++ b/parquet/src/arrow/buffer/offset_buffer.rs
@@ -18,6 +18,7 @@
 use crate::arrow::buffer::bit_util::iter_set_bits_rev;
 use crate::arrow::record_reader::buffer::ValuesBuffer;
 use crate::errors::{ParquetError, Result};
+use crate::util::utf8::check_valid_utf8;
 use arrow_array::{make_array, ArrayRef, OffsetSizeTrait};
 use arrow_buffer::{ArrowNativeType, Buffer};
 use arrow_data::ArrayDataBuilder;
@@ -117,14 +118,7 @@ impl<I: OffsetSizeTrait> OffsetBuffer<I> {
     ///
     /// [`Self::try_push`] can perform this validation check on insertion
     pub fn check_valid_utf8(&self, start_offset: usize) -> Result<()> {
-        match simdutf8::basic::from_utf8(&self.values.as_slice()[start_offset..]) {
-            Ok(_) => Ok(()),
-            Err(_) => {
-                let e = simdutf8::compat::from_utf8(&self.values.as_slice()[start_offset..])
-                    .unwrap_err();
-                Err(general_err!("encountered non UTF-8 data: {}", e))
-            }
-        }
+        check_valid_utf8(&self.values.as_slice()[start_offset..])
     }
 
     /// Converts this into an [`ArrayRef`] with the provided `data_type` and `null_buffer`
diff --git a/parquet/src/file/metadata/reader.rs b/parquet/src/file/metadata/reader.rs
index c6715a33b..d465a49c3 100644
--- a/parquet/src/file/metadata/reader.rs
+++ b/parquet/src/file/metadata/reader.rs
@@ -412,7 +412,9 @@ impl ParquetMetaDataReader {
         let bytes = match &remainder {
             Some((remainder_start, remainder)) if *remainder_start <= range.start => {
                 let offset = range.start - *remainder_start;
-                remainder.slice(offset..range.end - *remainder_start + offset)
+                let end = offset + range.end - range.start;
+                assert!(end <= remainder.len());
+                remainder.slice(offset..end)
             }
             // Note: this will potentially fetch data already in remainder, this keeps things simple
             _ => fetch.fetch(range.start..range.end).await?,
@@ -1052,5 +1054,41 @@ mod async_tests {
             .unwrap();
         assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
         assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
+
+        // Prefetch more than enough but less than the entire file
+        fetch_count.store(0, Ordering::SeqCst);
+        let f = MetadataFetchFn(&mut fetch);
+        let metadata = ParquetMetaDataReader::new()
+            .with_page_indexes(true)
+            .with_prefetch_hint(Some(len - 1000)) // prefetch entire file
+            .load_and_finish(f, len)
+            .await
+            .unwrap();
+        assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
+        assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
+
+        // Prefetch the entire file
+        fetch_count.store(0, Ordering::SeqCst);
+        let f = MetadataFetchFn(&mut fetch);
+        let metadata = ParquetMetaDataReader::new()
+            .with_page_indexes(true)
+            .with_prefetch_hint(Some(len)) // prefetch entire file
+            .load_and_finish(f, len)
+            .await
+            .unwrap();
+        assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
+        assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
+
+        // Prefetch more than the entire file
+        fetch_count.store(0, Ordering::SeqCst);
+        let f = MetadataFetchFn(&mut fetch);
+        let metadata = ParquetMetaDataReader::new()
+            .with_page_indexes(true)
+            .with_prefetch_hint(Some(len + 1000)) // prefetch entire file
+            .load_and_finish(f, len)
+            .await
+            .unwrap();
+        assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
+        assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
     }
 }
diff --git a/parquet/src/lib.rs b/parquet/src/lib.rs
index 3b63845e7..3ca0dbe98 100644
--- a/parquet/src/lib.rs
+++ b/parquet/src/lib.rs
@@ -131,6 +131,9 @@ pub mod data_type;
 pub use self::encodings::{decoding, encoding};
 
 experimental!(#[macro_use] mod util);
+
+pub use util::utf8;
+
 #[cfg(feature = "arrow")]
 pub mod arrow;
 pub mod column;
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index d9e9b22e8..68492e19f 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -1211,6 +1211,13 @@ fn from_thrift_helper(elements: &[SchemaElement], index: usize) -> Result<(usize
         ));
     }
     let element = &elements[index];
+
+    // Check for empty schema
+    if let (true, None | Some(0)) = (is_root_node, element.num_children) {
+        let builder = Type::group_type_builder(&element.name);
+        return Ok((index + 1, Arc::new(builder.build().unwrap())));
+    }
+
     let converted_type = ConvertedType::try_from(element.converted_type)?;
     // LogicalType is only present in v2 Parquet files. ConvertedType is always
     // populated, regardless of the version of the file (v1 or v2).
diff --git a/parquet/src/util/mod.rs b/parquet/src/util/mod.rs
index dfa1285af..143113247 100644
--- a/parquet/src/util/mod.rs
+++ b/parquet/src/util/mod.rs
@@ -19,8 +19,10 @@
 pub mod bit_util;
 mod bit_pack;
 pub(crate) mod interner;
+
 #[cfg(any(test, feature = "test_common"))]
 pub(crate) mod test_common;
+pub mod utf8;
 
 #[cfg(any(test, feature = "test_common"))]
 pub use self::test_common::page_util::{
diff --git a/parquet/src/util/utf8.rs b/parquet/src/util/utf8.rs
new file mode 100644
index 000000000..2a537b5e5
--- /dev/null
+++ b/parquet/src/util/utf8.rs
@@ -0,0 +1,57 @@
+// 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.
+
+//! [`check_valid_utf8`] validation function
+use crate::errors::{ParquetError, Result};
+
+/// Check that `val` is a valid UTF-8 sequence.
+///
+/// If the `simdutf8` feature is enabled, this function will use
+/// SIMD-accelerated validation from the [`simdutf8`] crate. Otherwise, it will use
+/// [`std::str::from_utf8`].
+///
+/// # Errors
+///
+/// Returns `Err::General` with a message compatible with [`std::str::from_utf8`] on failure.
+///
+/// # Example
+/// ```
+/// use parquet::utf8::check_valid_utf8;
+/// assert!(check_valid_utf8(b"hello").is_ok());
+/// assert!(check_valid_utf8(b"hello \xF0\x9F\x98\x8E").is_ok());
+/// // invalid UTF-8
+/// assert!(check_valid_utf8(b"hello \xF0\x9F\x98").is_err());
+/// ```
+///
+/// [`simdutf8`]: https://crates.io/crates/simdutf8
+#[inline(always)]
+pub fn check_valid_utf8(val: &[u8]) -> Result<()> {
+    #[cfg(feature = "simdutf8")]
+    match simdutf8::basic::from_utf8(val) {
+        Ok(_) => Ok(()),
+        Err(_) => {
+            // Use simdutf8::compat to return details about the decoding error
+            let e = simdutf8::compat::from_utf8(val).unwrap_err();
+            Err(general_err!("encountered non UTF-8 data: {}", e))
+        }
+    }
+    #[cfg(not(feature = "simdutf8"))]
+    match std::str::from_utf8(val) {
+        Ok(_) => Ok(()),
+        Err(e) => Err(general_err!("encountered non UTF-8 data: {}", e)),
+    }
+}


Test patch:
diff --git a/object_store/tests/get_range_file.rs b/object_store/tests/get_range_file.rs
index e500fc8ac..6790c11e7 100644
--- a/object_store/tests/get_range_file.rs
+++ b/object_store/tests/get_range_file.rs
@@ -90,12 +90,15 @@ async fn test_get_range() {
     let fetched = store.get(&path).await.unwrap().bytes().await.unwrap();
     assert_eq!(expected, fetched);
 
-    for range in [0..10, 3..5, 0..expected.len()] {
+    for range in [0..10, 3..5, 0..expected.len() as u64] {
         let data = store.get_range(&path, range.clone()).await.unwrap();
-        assert_eq!(&data[..], &expected[range])
+        assert_eq!(
+            &data[..],
+            &expected[range.start as usize..range.end as usize]
+        )
     }
 
-    let over_range = 0..(expected.len() * 2);
+    let over_range = 0..(expected.len() as u64 * 2);
     let data = store.get_range(&path, over_range.clone()).await.unwrap();
     assert_eq!(&data[..], expected)
 }
@@ -113,10 +116,10 @@ async fn test_get_opts_over_range() {
     store.put(&path, expected.clone().into()).await.unwrap();
 
     let opts = GetOptions {
-        range: Some(GetRange::Bounded(0..(expected.len() * 2))),
+        range: Some(GetRange::Bounded(0..(expected.len() as u64 * 2))),
         ..Default::default()
     };
     let res = store.get_opts(&path, opts).await.unwrap();
-    assert_eq!(res.range, 0..expected.len());
+    assert_eq!(res.range, 0..expected.len() as u64);
     assert_eq!(res.bytes().await.unwrap(), expected);
 }