Repository: apache/arrow-rs
Pull request #6382 (2024-09-11 13:28:11)

object_score: Support Azure Fabric OAuth Provider
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes #.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

In Azure Fabric, we use token service to get user access token, for supporting long reading and writing operation and auto refresh access token, we implement this.

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->


This pull request introduces significant enhancements to the Azure integration within the `object_store` module, including the implementation of a new `FabricTokenOAuthProvider`. These changes aim to improve the authentication mechanism and add support for fabric token services.

### Azure Builder Enhancements:
* Added new fields to `MicrosoftAzureBuilder` to support fabric token services, including `fabric_token_service_url`, `fabric_workload_host`, `fabric_session_token`, and `fabric_cluster_identifier`. (`object_store/src/azure/builder.rs` [object_store/src/azure/builder.rsR175-R182](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R175-R182))
* Updated `AzureConfigKey` to include new configuration keys related to fabric token services. (`object_store/src/azure/builder.rs` [object_store/src/azure/builder.rsR347-R374](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R347-R374))
* Modified `impl AsRef<str>` and `impl FromStr` for `AzureConfigKey` to handle new fabric token service keys. (`object_store/src/azure/builder.rs` [[1]](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R400-R403) [[2]](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R449-R456)
* Enhanced `MicrosoftAzureBuilder` to set and get the new fabric token service-related fields. (`object_store/src/azure/builder.rs` [[1]](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R576-R583) [[2]](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R620-R623)
* Added logic to `MicrosoftAzureBuilder` to create a `FabricTokenOAuthProvider` if fabric token service fields are provided. (`object_store/src/azure/builder.rs` [object_store/src/azure/builder.rsR919-R942](diffhunk://#diff-8a210e33eba6ee7a3f8413616b7737e5f50658e481e1707024ab91b97e106363R919-R942))

### Credential Enhancements:
* Introduced `FabricTokenOAuthProvider` to handle OAuth token challenges for fabric token services, including methods for token validation and fetching. (`object_store/src/azure/credential.rs` [object_store/src/azure/credential.rsR943-R1049](diffhunk://#diff-53498fbed9bb6cfcf50c4c9335b23cd5add9d77be3a150dca7235a05716718b4R943-R1049))
* Added new headers and constants related to fabric token services. (`object_store/src/azure/credential.rs` [object_store/src/azure/credential.rsR55-R63](diffhunk://#diff-53498fbed9bb6cfcf50c4c9335b23cd5add9d77be3a150dca7235a05716718b4R55-R63))

These changes collectively enhance the Azure integration by supporting more complex authentication mechanisms, particularly for environments utilizing fabric token services.

# Are there any user-facing changes?

After that, we can set some environment variables to make it auto refresh access token in Fabric Notebook.
```
# For Fabric Spark Notebook
import os
import urllib.parse
workload_endpoint = urllib.parse.urlparse(f"{spark.conf.get('trident.lakehouse.tokenservice.endpoint')}/access")
os.environ['azure_fabric_token_service_url'.upper()] = f"https://{spark.conf.get('spark.tokenServiceEndpoint')}/api/v1/proxy{workload_endpoint.path}"
os.environ['azure_fabric_workload_host'.upper()] = f"{workload_endpoint.scheme}://{workload_endpoint.hostname}"
os.environ['azure_fabric_session_token'.upper()] = spark.conf.get("trident.session.token")
os.environ['azure_fabric_cluster_identifier'.upper()] = spark.conf.get("spark.synapse.clusteridentifier")
os.environ['azure_storage_token'.upper()] = notebookutils.credentials.getToken("storage")

# For Fabric Python Notebook
import os
from notebookutils.common import configs
import urllib.parse
workload_endpoint = urllib.parse.urlparse(f"{configs.workload_endpoint()}/access")
os.environ['azure_fabric_token_service_url'.upper()] = f"{configs.ts_endpoint()}/api/v1/proxy{workload_endpoint.path}"
os.environ['azure_fabric_workload_host'.upper()] = f"{workload_endpoint.scheme}://{workload_endpoint.hostname}"
os.environ['azure_fabric_session_token'.upper()] = configs.session_token()
os.environ['azure_fabric_cluster_identifier'.upper()] = configs.cluster_identifier()
os.environ['azure_storage_token'.upper()] = notebookutils.credentials.getToken("storage")
```

Then, user can read/write delta table without storage_option.
```
from deltalake import DeltaTable
dt = DeltaTable('abfss://xxxx@onelake.dfs.fabric.microsoft.com/LH.Lakehouse/Tables/dbo/test')
df = dt.to_pyarrow_dataset().head(10).to_pandas()
```

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 1937fafe3..900f0fe83 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -48,7 +48,6 @@ on:
       - arrow/**
 
 jobs:
-
   integration:
     name: Archery test With other arrows
     runs-on: ubuntu-latest
@@ -65,6 +64,7 @@ jobs:
       ARROW_INTEGRATION_GO: ON
       ARROW_INTEGRATION_JAVA: ON
       ARROW_INTEGRATION_JS: ON
+      ARCHERY_INTEGRATION_TARGET_IMPLEMENTATIONS: "rust"
       ARCHERY_INTEGRATION_WITH_NANOARROW: "1"
       # https://github.com/apache/arrow/pull/38403/files#r1371281630
       ARCHERY_INTEGRATION_WITH_RUST: "1"
@@ -118,9 +118,9 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        rust: [ stable ]
-        # PyArrow 13 was the last version prior to introduction to Arrow PyCapsules
-        pyarrow: [ "13", "14" ]
+        rust: [stable]
+        # PyArrow 15 was the first version to introduce StringView/BinaryView support
+        pyarrow: ["15", "16", "17"]
     steps:
       - uses: actions/checkout@v4
         with:
diff --git a/arrow-arith/src/arity.rs b/arrow-arith/src/arity.rs
index 17c1b0dbc..bb983e122 100644
--- a/arrow-arith/src/arity.rs
+++ b/arrow-arith/src/arity.rs
@@ -313,8 +313,6 @@ where
         ))));
     }
 
-    let nulls = NullBuffer::union(a.logical_nulls().as_ref(), b.logical_nulls().as_ref());
-
     let mut builder = a.into_builder()?;
 
     builder
@@ -323,14 +321,21 @@ where
         .zip(b.values())
         .for_each(|(l, r)| *l = op(*l, *r));
 
-    let array_builder = builder.finish().into_data().into_builder().nulls(nulls);
+    let array = builder.finish();
+
+    // The builder has the null buffer from `a`, it is not changed.
+    let nulls = NullBuffer::union(array.logical_nulls().as_ref(), b.logical_nulls().as_ref());
+
+    let array_builder = array.into_data().into_builder().nulls(nulls);
 
     let array_data = unsafe { array_builder.build_unchecked() };
     Ok(Ok(PrimitiveArray::<T>::from(array_data)))
 }
 
-/// Applies the provided fallible binary operation across `a` and `b`, returning any error,
-/// and collecting the results into a [`PrimitiveArray`]. If any index is null in either `a`
+/// Applies the provided fallible binary operation across `a` and `b`.
+///
+/// This will return any error encountered, or collect the results into
+/// a [`PrimitiveArray`]. If any index is null in either `a`
 /// or `b`, the corresponding index in the result will also be null
 ///
 /// Like [`try_unary`] the function is only evaluated for non-null indices
@@ -381,12 +386,15 @@ where
 }
 
 /// Applies the provided fallible binary operation across `a` and `b` by mutating the mutable
-/// [`PrimitiveArray`] `a` with the results, returning any error. If any index is null in
-/// either `a` or `b`, the corresponding index in the result will also be null
+/// [`PrimitiveArray`] `a` with the results.
 ///
-/// Like [`try_unary`] the function is only evaluated for non-null indices
+/// Returns any error encountered, or collects the results into a [`PrimitiveArray`] as return
+/// value. If any index is null in either `a` or `b`, the corresponding index in the result will
+/// also be null.
+///
+/// Like [`try_unary`] the function is only evaluated for non-null indices.
 ///
-/// See [`binary_mut`] for errors and buffer reuse information
+/// See [`binary_mut`] for errors and buffer reuse information.
 pub fn try_binary_mut<T, F>(
     a: PrimitiveArray<T>,
     b: &PrimitiveArray<T>,
@@ -413,7 +421,8 @@ where
         try_binary_no_nulls_mut(len, a, b, op)
     } else {
         let nulls =
-            NullBuffer::union(a.logical_nulls().as_ref(), b.logical_nulls().as_ref()).unwrap();
+            create_union_null_buffer(a.logical_nulls().as_ref(), b.logical_nulls().as_ref())
+                .unwrap();
 
         let mut builder = a.into_builder()?;
 
@@ -435,6 +444,22 @@ where
     }
 }
 
+/// Computes the union of the nulls in two optional [`NullBuffer`] which
+/// is not shared with the input buffers.
+///
+/// The union of the nulls is the same as `NullBuffer::union(lhs, rhs)` but
+/// it does not increase the reference count of the null buffer.
+fn create_union_null_buffer(
+    lhs: Option<&NullBuffer>,
+    rhs: Option<&NullBuffer>,
+) -> Option<NullBuffer> {
+    match (lhs, rhs) {
+        (Some(lhs), Some(rhs)) => Some(NullBuffer::new(lhs.inner() & rhs.inner())),
+        (Some(n), None) | (None, Some(n)) => Some(NullBuffer::new(n.inner() & n.inner())),
+        (None, None) => None,
+    }
+}
+
 /// This intentional inline(never) attribute helps LLVM optimize the loop.
 #[inline(never)]
 fn try_binary_no_nulls<A: ArrayAccessor, B: ArrayAccessor, F, O>(
@@ -557,6 +582,25 @@ mod tests {
         assert_eq!(c, expected);
     }
 
+    #[test]
+    fn test_binary_mut_null_buffer() {
+        let a = Int32Array::from(vec![Some(3), Some(4), Some(5), Some(6), None]);
+
+        let b = Int32Array::from(vec![Some(10), Some(11), Some(12), Some(13), Some(14)]);
+
+        let r1 = binary_mut(a, &b, |a, b| a + b).unwrap();
+
+        let a = Int32Array::from(vec![Some(3), Some(4), Some(5), Some(6), None]);
+        let b = Int32Array::new(
+            vec![10, 11, 12, 13, 14].into(),
+            Some(vec![true, true, true, true, true].into()),
+        );
+
+        // unwrap here means that no copying occured
+        let r2 = binary_mut(a, &b, |a, b| a + b).unwrap();
+        assert_eq!(r1.unwrap(), r2.unwrap());
+    }
+
     #[test]
     fn test_try_binary_mut() {
         let a = Int32Array::from(vec![15, 14, 9, 8, 1]);
@@ -587,6 +631,25 @@ mod tests {
         .expect_err("should got error");
     }
 
+    #[test]
+    fn test_try_binary_mut_null_buffer() {
+        let a = Int32Array::from(vec![Some(3), Some(4), Some(5), Some(6), None]);
+
+        let b = Int32Array::from(vec![Some(10), Some(11), Some(12), Some(13), Some(14)]);
+
+        let r1 = try_binary_mut(a, &b, |a, b| Ok(a + b)).unwrap();
+
+        let a = Int32Array::from(vec![Some(3), Some(4), Some(5), Some(6), None]);
+        let b = Int32Array::new(
+            vec![10, 11, 12, 13, 14].into(),
+            Some(vec![true, true, true, true, true].into()),
+        );
+
+        // unwrap here means that no copying occured
+        let r2 = try_binary_mut(a, &b, |a, b| Ok(a + b)).unwrap();
+        assert_eq!(r1.unwrap(), r2.unwrap());
+    }
+
     #[test]
     fn test_unary_dict_mut() {
         let values = Int32Array::from(vec![Some(10), Some(20), None]);
diff --git a/arrow-arith/src/bitwise.rs b/arrow-arith/src/bitwise.rs
index c829a3c29..a3c18136c 100644
--- a/arrow-arith/src/bitwise.rs
+++ b/arrow-arith/src/bitwise.rs
@@ -15,6 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Module contains bitwise operations on arrays
+
 use crate::arity::{binary, unary};
 use arrow_array::*;
 use arrow_buffer::ArrowNativeType;
diff --git a/arrow-arith/src/lib.rs b/arrow-arith/src/lib.rs
index 2d5451e04..c8b6412e5 100644
--- a/arrow-arith/src/lib.rs
+++ b/arrow-arith/src/lib.rs
@@ -17,6 +17,7 @@
 
 //! Arrow arithmetic and aggregation kernels
 
+#![warn(missing_docs)]
 pub mod aggregate;
 #[doc(hidden)] // Kernels to be removed in a future release
 pub mod arithmetic;
diff --git a/arrow-arith/src/temporal.rs b/arrow-arith/src/temporal.rs
index 5f3eeb325..09d690d32 100644
--- a/arrow-arith/src/temporal.rs
+++ b/arrow-arith/src/temporal.rs
@@ -666,6 +666,7 @@ impl<T: Datelike> ChronoDateExt for T {
 
 /// Parse the given string into a string representing fixed-offset that is correct as of the given
 /// UTC NaiveDateTime.
+///
 /// Note that the offset is function of time and can vary depending on whether daylight savings is
 /// in effect or not. e.g. Australia/Sydney is +10:00 or +11:00 depending on DST.
 #[deprecated(note = "Use arrow_array::timezone::Tz instead")]
@@ -811,6 +812,7 @@ where
 }
 
 /// Extracts the day of a given temporal array as an array of integers.
+///
 /// If the given array isn't temporal primitive or dictionary array,
 /// an `Err` will be returned.
 #[deprecated(since = "51.0.0", note = "Use `date_part` instead")]
@@ -828,7 +830,8 @@ where
     date_part_primitive(array, DatePart::Day)
 }
 
-/// Extracts the day of year of a given temporal array as an array of integers
+/// Extracts the day of year of a given temporal array as an array of integers.
+///
 /// The day of year that ranges from 1 to 366.
 /// If the given array isn't temporal primitive or dictionary array,
 /// an `Err` will be returned.
@@ -837,7 +840,8 @@ pub fn doy_dyn(array: &dyn Array) -> Result<ArrayRef, ArrowError> {
     date_part(array, DatePart::DayOfYear)
 }
 
-/// Extracts the day of year of a given temporal primitive array as an array of integers
+/// Extracts the day of year of a given temporal primitive array as an array of integers.
+///
 /// The day of year that ranges from 1 to 366
 #[deprecated(since = "51.0.0", note = "Use `date_part` instead")]
 pub fn doy<T>(array: &PrimitiveArray<T>) -> Result<Int32Array, ArrowError>
diff --git a/arrow-array/src/array/binary_array.rs b/arrow-array/src/array/binary_array.rs
index 6b18cbc2d..8f8a39b20 100644
--- a/arrow-array/src/array/binary_array.rs
+++ b/arrow-array/src/array/binary_array.rs
@@ -84,7 +84,7 @@ impl<OffsetSize: OffsetSizeTrait> GenericBinaryArray<OffsetSize> {
     pub fn take_iter<'a>(
         &'a self,
         indexes: impl Iterator<Item = Option<usize>> + 'a,
-    ) -> impl Iterator<Item = Option<&[u8]>> + 'a {
+    ) -> impl Iterator<Item = Option<&'a [u8]>> {
         indexes.map(|opt_index| opt_index.map(|index| self.value(index)))
     }
 
@@ -95,7 +95,7 @@ impl<OffsetSize: OffsetSizeTrait> GenericBinaryArray<OffsetSize> {
     pub unsafe fn take_iter_unchecked<'a>(
         &'a self,
         indexes: impl Iterator<Item = Option<usize>> + 'a,
-    ) -> impl Iterator<Item = Option<&[u8]>> + 'a {
+    ) -> impl Iterator<Item = Option<&'a [u8]>> {
         indexes.map(|opt_index| opt_index.map(|index| self.value_unchecked(index)))
     }
 }
diff --git a/arrow-array/src/array/string_array.rs b/arrow-array/src/array/string_array.rs
index 4722de55d..25581cfaa 100644
--- a/arrow-array/src/array/string_array.rs
+++ b/arrow-array/src/array/string_array.rs
@@ -42,7 +42,7 @@ impl<OffsetSize: OffsetSizeTrait> GenericStringArray<OffsetSize> {
     pub fn take_iter<'a>(
         &'a self,
         indexes: impl Iterator<Item = Option<usize>> + 'a,
-    ) -> impl Iterator<Item = Option<&str>> + 'a {
+    ) -> impl Iterator<Item = Option<&'a str>> {
         indexes.map(|opt_index| opt_index.map(|index| self.value(index)))
     }
 
@@ -53,7 +53,7 @@ impl<OffsetSize: OffsetSizeTrait> GenericStringArray<OffsetSize> {
     pub unsafe fn take_iter_unchecked<'a>(
         &'a self,
         indexes: impl Iterator<Item = Option<usize>> + 'a,
-    ) -> impl Iterator<Item = Option<&str>> + 'a {
+    ) -> impl Iterator<Item = Option<&'a str>> {
         indexes.map(|opt_index| opt_index.map(|index| self.value_unchecked(index)))
     }
 
diff --git a/arrow-array/src/builder/generic_bytes_builder.rs b/arrow-array/src/builder/generic_bytes_builder.rs
index 516f45be3..e85a2fed0 100644
--- a/arrow-array/src/builder/generic_bytes_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_builder.rs
@@ -60,8 +60,9 @@ impl<T: ByteArrayType> GenericByteBuilder<T> {
     /// Creates a new  [`GenericByteBuilder`] from buffers.
     ///
     /// # Safety
-    /// This doesn't verify buffer contents as it assumes the buffers are from existing and
-    /// valid [`GenericByteArray`].
+    ///
+    /// This doesn't verify buffer contents as it assumes the buffers are from
+    /// existing and valid [`GenericByteArray`].
     pub unsafe fn new_from_buffer(
         offsets_buffer: MutableBuffer,
         value_buffer: MutableBuffer,
@@ -88,9 +89,19 @@ impl<T: ByteArrayType> GenericByteBuilder<T> {
 
     /// Appends a value into the builder.
     ///
+    /// See the [GenericStringBuilder] documentation for examples of
+    /// incrementally building string values with multiple `write!` calls.
+    ///
     /// # Panics
     ///
-    /// Panics if the resulting length of [`Self::values_slice`] would exceed `T::Offset::MAX`
+    /// Panics if the resulting length of [`Self::values_slice`] would exceed
+    /// `T::Offset::MAX` bytes.
+    ///
+    /// For example, this can happen with [`StringArray`] or [`BinaryArray`]
+    /// where the total length of all values exceeds 2GB
+    ///
+    /// [`StringArray`]: crate::StringArray
+    /// [`BinaryArray`]: crate::BinaryArray
     #[inline]
     pub fn append_value(&mut self, value: impl AsRef<T::Native>) {
         self.value_builder.append_slice(value.as_ref().as_ref());
@@ -99,6 +110,11 @@ impl<T: ByteArrayType> GenericByteBuilder<T> {
     }
 
     /// Append an `Option` value into the builder.
+    ///
+    /// - A `None` value will append a null value.
+    /// - A `Some` value will append the value.
+    ///
+    /// See [`Self::append_value`] for more panic information.
     #[inline]
     pub fn append_option(&mut self, value: Option<impl AsRef<T::Native>>) {
         match value {
@@ -227,30 +243,51 @@ impl<T: ByteArrayType, V: AsRef<T::Native>> Extend<Option<V>> for GenericByteBui
 /// Values can be appended using [`GenericByteBuilder::append_value`], and nulls with
 /// [`GenericByteBuilder::append_null`].
 ///
-/// Additionally, implements [`std::fmt::Write`] with any written data included in the next
-/// appended value. This allows use with [`std::fmt::Display`] without intermediate allocations
+/// This builder also implements [`std::fmt::Write`] with any written data
+/// included in the next appended value. This allows using [`std::fmt::Display`]
+/// with standard Rust idioms like `write!` and `writeln!` to write data
+/// directly to the builder without intermediate allocations.
+///
+/// # Example writing strings with `append_value`
+/// ```
+/// # use arrow_array::builder::GenericStringBuilder;
+/// let mut builder = GenericStringBuilder::<i32>::new();
+///
+/// // Write one string value
+/// builder.append_value("foobarbaz");
+///
+/// // Write a second string
+/// builder.append_value("v2");
+///
+/// let array = builder.finish();
+/// assert_eq!(array.value(0), "foobarbaz");
+/// assert_eq!(array.value(1), "v2");
+/// ```
+///
+/// # Example incrementally writing strings with `std::fmt::Write`
 ///
-/// # Example
 /// ```
 /// # use std::fmt::Write;
 /// # use arrow_array::builder::GenericStringBuilder;
 /// let mut builder = GenericStringBuilder::<i32>::new();
 ///
-/// // Write data
+/// // Write data in multiple `write!` calls
 /// write!(builder, "foo").unwrap();
 /// write!(builder, "bar").unwrap();
-///
-/// // Finish value
+/// // The next call to append_value finishes the current string
+/// // including all previously written strings.
 /// builder.append_value("baz");
 ///
-/// // Write second value
+/// // Write second value with a single write call
 /// write!(builder, "v2").unwrap();
+/// // finish the value by calling append_value with an empty string
 /// builder.append_value("");
 ///
 /// let array = builder.finish();
 /// assert_eq!(array.value(0), "foobarbaz");
 /// assert_eq!(array.value(1), "v2");
 /// ```
+///
 pub type GenericStringBuilder<O> = GenericByteBuilder<GenericStringType<O>>;
 
 impl<O: OffsetSizeTrait> Write for GenericStringBuilder<O> {
diff --git a/arrow-array/src/builder/generic_bytes_view_builder.rs b/arrow-array/src/builder/generic_bytes_view_builder.rs
index deaf447d0..09277c679 100644
--- a/arrow-array/src/builder/generic_bytes_view_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_view_builder.rs
@@ -368,6 +368,9 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
         let len = self.views_builder.len();
         let views = ScalarBuffer::new(self.views_builder.finish(), 0, len);
         let nulls = self.null_buffer_builder.finish();
+        if let Some((ref mut ht, _)) = self.string_tracker.as_mut() {
+            ht.clear();
+        }
         // SAFETY: valid by construction
         unsafe { GenericByteViewArray::new_unchecked(views, completed, nulls) }
     }
@@ -512,7 +515,8 @@ fn make_inlined_view<const LEN: usize>(data: &[u8]) -> u128 {
     u128::from_le_bytes(view_buffer)
 }
 
-/// Create a view based on the given data, block id and offset
+/// Create a view based on the given data, block id and offset.
+///
 /// Note that the code below is carefully examined with x86_64 assembly code: <https://godbolt.org/z/685YPsd5G>
 /// The goal is to avoid calling into `ptr::copy_non_interleave`, which makes function call (i.e., not inlined),
 /// which slows down things.
@@ -590,6 +594,20 @@ mod tests {
         assert_eq!(array.views().get(1), array.views().get(5));
     }
 
+    #[test]
+    fn test_string_view_deduplicate_after_finish() {
+        let mut builder = StringViewBuilder::new().with_deduplicate_strings();
+
+        let value_1 = "long string to test string view";
+        let value_2 = "not so similar string but long";
+        builder.append_value(value_1);
+        let _array = builder.finish();
+        builder.append_value(value_2);
+        let _array = builder.finish();
+        builder.append_value(value_1);
+        let _array = builder.finish();
+    }
+
     #[test]
     fn test_string_view() {
         let b1 = Buffer::from(b"world\xFFbananas\xF0\x9F\x98\x81");
diff --git a/arrow-array/src/ffi.rs b/arrow-array/src/ffi.rs
index 1d76ed62d..a28b3f746 100644
--- a/arrow-array/src/ffi.rs
+++ b/arrow-array/src/ffi.rs
@@ -193,6 +193,13 @@ fn bit_width(data_type: &DataType, i: usize) -> Result<usize> {
                 "The datatype \"{data_type:?}\" expects 3 buffers, but requested {i}. Please verify that the C data interface is correctly implemented."
             )))
         }
+        // Variable-sized views: have 3 or more buffers.
+        // Buffer 1 are the u128 views
+        // Buffers 2...N-1 are u8 byte buffers
+        (DataType::Utf8View, 1) | (DataType::BinaryView,1) => u128::BITS as _,
+        (DataType::Utf8View, _) | (DataType::BinaryView, _) => {
+            u8::BITS as _
+        }
         // type ids. UnionArray doesn't have null bitmap so buffer index begins with 0.
         (DataType::Union(_, _), 0) => i8::BITS as _,
         // Only DenseUnion has 2nd buffer
@@ -300,7 +307,7 @@ impl<'a> ImportedArrowArray<'a> {
         };
 
         let data_layout = layout(&self.data_type);
-        let buffers = self.buffers(data_layout.can_contain_null_mask)?;
+        let buffers = self.buffers(data_layout.can_contain_null_mask, data_layout.variadic)?;
 
         let null_bit_buffer = if data_layout.can_contain_null_mask {
             self.null_bit_buffer()
@@ -373,13 +380,30 @@ impl<'a> ImportedArrowArray<'a> {
 
     /// returns all buffers, as organized by Rust (i.e. null buffer is skipped if it's present
     /// in the spec of the type)
-    fn buffers(&self, can_contain_null_mask: bool) -> Result<Vec<Buffer>> {
+    fn buffers(&self, can_contain_null_mask: bool, variadic: bool) -> Result<Vec<Buffer>> {
         // + 1: skip null buffer
         let buffer_begin = can_contain_null_mask as usize;
-        (buffer_begin..self.array.num_buffers())
-            .map(|index| {
-                let len = self.buffer_len(index, &self.data_type)?;
+        let buffer_end = self.array.num_buffers() - usize::from(variadic);
+
+        let variadic_buffer_lens = if variadic {
+            // Each views array has 1 (optional) null buffer, 1 views buffer, 1 lengths buffer.
+            // Rest are variadic.
+            let num_variadic_buffers =
+                self.array.num_buffers() - (2 + usize::from(can_contain_null_mask));
+            if num_variadic_buffers == 0 {
+                &[]
+            } else {
+                let lengths = self.array.buffer(self.array.num_buffers() - 1);
+                // SAFETY: is lengths is non-null, then it must be valid for up to num_variadic_buffers.
+                unsafe { std::slice::from_raw_parts(lengths.cast::<i64>(), num_variadic_buffers) }
+            }
+        } else {
+            &[]
+        };
 
+        (buffer_begin..buffer_end)
+            .map(|index| {
+                let len = self.buffer_len(index, variadic_buffer_lens, &self.data_type)?;
                 match unsafe { create_buffer(self.owner.clone(), self.array, index, len) } {
                     Some(buf) => Ok(buf),
                     None if len == 0 => {
@@ -399,7 +423,12 @@ impl<'a> ImportedArrowArray<'a> {
     /// Rust implementation uses fixed-sized buffers, which require knowledge of their `len`.
     /// for variable-sized buffers, such as the second buffer of a stringArray, we need
     /// to fetch offset buffer's len to build the second buffer.
-    fn buffer_len(&self, i: usize, dt: &DataType) -> Result<usize> {
+    fn buffer_len(
+        &self,
+        i: usize,
+        variadic_buffer_lengths: &[i64],
+        dt: &DataType,
+    ) -> Result<usize> {
         // Special handling for dictionary type as we only care about the key type in the case.
         let data_type = match dt {
             DataType::Dictionary(key_data_type, _) => key_data_type.as_ref(),
@@ -430,7 +459,7 @@ impl<'a> ImportedArrowArray<'a> {
                 }
 
                 // the len of the data buffer (buffer 2) equals the last value of the offset buffer (buffer 1)
-                let len = self.buffer_len(1, dt)?;
+                let len = self.buffer_len(1, variadic_buffer_lengths, dt)?;
                 // first buffer is the null buffer => add(1)
                 // we assume that pointer is aligned for `i32`, as Utf8 uses `i32` offsets.
                 #[allow(clippy::cast_ptr_alignment)]
@@ -444,7 +473,7 @@ impl<'a> ImportedArrowArray<'a> {
                 }
 
                 // the len of the data buffer (buffer 2) equals the last value of the offset buffer (buffer 1)
-                let len = self.buffer_len(1, dt)?;
+                let len = self.buffer_len(1, variadic_buffer_lengths, dt)?;
                 // first buffer is the null buffer => add(1)
                 // we assume that pointer is aligned for `i64`, as Large uses `i64` offsets.
                 #[allow(clippy::cast_ptr_alignment)]
@@ -452,6 +481,16 @@ impl<'a> ImportedArrowArray<'a> {
                 // get last offset
                 (unsafe { *offset_buffer.add(len / size_of::<i64>() - 1) }) as usize
             }
+            // View types: these have variadic buffers.
+            // Buffer 1 is the views buffer, which stores 1 u128 per length of the array.
+            // Buffers 2..N-1 are the buffers holding the byte data. Their lengths are variable.
+            // Buffer N is of length (N - 2) and stores i64 containing the lengths of buffers 2..N-1
+            (DataType::Utf8View, 1) | (DataType::BinaryView, 1) => {
+                std::mem::size_of::<u128>() * length
+            }
+            (DataType::Utf8View, i) | (DataType::BinaryView, i) => {
+                variadic_buffer_lengths[i - 2] as usize
+            }
             // buffer len of primitive types
             _ => {
                 let bits = bit_width(data_type, i)?;
@@ -1229,18 +1268,18 @@ mod tests_from_ffi {
     use arrow_data::ArrayData;
     use arrow_schema::{DataType, Field};
 
-    use crate::types::Int32Type;
+    use super::{ImportedArrowArray, Result};
+    use crate::builder::GenericByteViewBuilder;
+    use crate::types::{BinaryViewType, ByteViewType, Int32Type, StringViewType};
     use crate::{
         array::{
             Array, BooleanArray, DictionaryArray, FixedSizeBinaryArray, FixedSizeListArray,
             Int32Array, Int64Array, StringArray, StructArray, UInt32Array, UInt64Array,
         },
         ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema},
-        make_array, ArrayRef, ListArray,
+        make_array, ArrayRef, GenericByteViewArray, ListArray,
     };
 
-    use super::{ImportedArrowArray, Result};
-
     fn test_round_trip(expected: &ArrayData) -> Result<()> {
         // here we export the array
         let array = FFI_ArrowArray::new(expected);
@@ -1453,8 +1492,8 @@ mod tests_from_ffi {
             owner: &array,
         };
 
-        let offset_buf_len = imported_array.buffer_len(1, &imported_array.data_type)?;
-        let data_buf_len = imported_array.buffer_len(2, &imported_array.data_type)?;
+        let offset_buf_len = imported_array.buffer_len(1, &[], &imported_array.data_type)?;
+        let data_buf_len = imported_array.buffer_len(2, &[], &imported_array.data_type)?;
 
         assert_eq!(offset_buf_len, 4);
         assert_eq!(data_buf_len, 0);
@@ -1472,6 +1511,18 @@ mod tests_from_ffi {
         StringArray::from(array)
     }
 
+    fn roundtrip_byte_view_array<T: ByteViewType>(
+        array: GenericByteViewArray<T>,
+    ) -> GenericByteViewArray<T> {
+        let data = array.into_data();
+
+        let array = FFI_ArrowArray::new(&data);
+        let schema = FFI_ArrowSchema::try_from(data.data_type()).unwrap();
+
+        let array = unsafe { from_ffi(array, &schema) }.unwrap();
+        GenericByteViewArray::<T>::from(array)
+    }
+
     fn extend_array(array: &dyn Array) -> ArrayRef {
         let len = array.len();
         let data = array.to_data();
@@ -1551,4 +1602,93 @@ mod tests_from_ffi {
             &imported
         );
     }
+
+    /// Helper trait to allow us to use easily strings as either BinaryViewType::Native or
+    /// StringViewType::Native scalars.
+    trait NativeFromStr {
+        fn from_str(value: &str) -> &Self;
+    }
+
+    impl NativeFromStr for str {
+        fn from_str(value: &str) -> &Self {
+            value
+        }
+    }
+
+    impl NativeFromStr for [u8] {
+        fn from_str(value: &str) -> &Self {
+            value.as_bytes()
+        }
+    }
+
+    #[test]
+    fn test_round_trip_byte_view() {
+        fn test_case<T>()
+        where
+            T: ByteViewType,
+            T::Native: NativeFromStr,
+        {
+            macro_rules! run_test_case {
+                ($array:expr) => {{
+                    // round-trip through C  Data Interface
+                    let len = $array.len();
+                    let imported = roundtrip_byte_view_array($array);
+                    assert_eq!(imported.len(), len);
+
+                    let copied = extend_array(&imported);
+                    assert_eq!(
+                        copied
+                            .as_any()
+                            .downcast_ref::<GenericByteViewArray<T>>()
+                            .unwrap(),
+                        &imported
+                    );
+                }};
+            }
+
+            // Empty test case.
+            let empty = GenericByteViewBuilder::<T>::new().finish();
+            run_test_case!(empty);
+
+            // All inlined strings test case.
+            let mut all_inlined = GenericByteViewBuilder::<T>::new();
+            all_inlined.append_value(T::Native::from_str("inlined1"));
+            all_inlined.append_value(T::Native::from_str("inlined2"));
+            all_inlined.append_value(T::Native::from_str("inlined3"));
+            let all_inlined = all_inlined.finish();
+            assert_eq!(all_inlined.data_buffers().len(), 0);
+            run_test_case!(all_inlined);
+
+            // some inlined + non-inlined, 1 variadic buffer.
+            let mixed_one_variadic = {
+                let mut builder = GenericByteViewBuilder::<T>::new();
+                builder.append_value(T::Native::from_str("inlined"));
+                let block_id =
+                    builder.append_block(Buffer::from("non-inlined-string-buffer".as_bytes()));
+                builder.try_append_view(block_id, 0, 25).unwrap();
+                builder.finish()
+            };
+            assert_eq!(mixed_one_variadic.data_buffers().len(), 1);
+            run_test_case!(mixed_one_variadic);
+
+            // inlined + non-inlined, 2 variadic buffers.
+            let mixed_two_variadic = {
+                let mut builder = GenericByteViewBuilder::<T>::new();
+                builder.append_value(T::Native::from_str("inlined"));
+                let block_id =
+                    builder.append_block(Buffer::from("non-inlined-string-buffer".as_bytes()));
+                builder.try_append_view(block_id, 0, 25).unwrap();
+
+                let block_id = builder
+                    .append_block(Buffer::from("another-non-inlined-string-buffer".as_bytes()));
+                builder.try_append_view(block_id, 0, 33).unwrap();
+                builder.finish()
+            };
+            assert_eq!(mixed_two_variadic.data_buffers().len(), 2);
+            run_test_case!(mixed_two_variadic);
+        }
+
+        test_case::<StringViewType>();
+        test_case::<BinaryViewType>();
+    }
 }
diff --git a/arrow-array/src/ffi_stream.rs b/arrow-array/src/ffi_stream.rs
index 6f3405ead..34f0cd7cf 100644
--- a/arrow-array/src/ffi_stream.rs
+++ b/arrow-array/src/ffi_stream.rs
@@ -83,17 +83,18 @@ const ENOSYS: i32 = 78;
 /// This was created by bindgen
 #[repr(C)]
 #[derive(Debug)]
-#[allow(missing_docs)]
+#[allow(non_camel_case_types)]
 pub struct FFI_ArrowArrayStream {
-    pub get_schema: Option<
-        unsafe extern "C" fn(arg1: *mut FFI_ArrowArrayStream, out: *mut FFI_ArrowSchema) -> c_int,
-    >,
-    pub get_next: Option<
-        unsafe extern "C" fn(arg1: *mut FFI_ArrowArrayStream, out: *mut FFI_ArrowArray) -> c_int,
-    >,
-    pub get_last_error:
-        Option<unsafe extern "C" fn(arg1: *mut FFI_ArrowArrayStream) -> *const c_char>,
-    pub release: Option<unsafe extern "C" fn(arg1: *mut FFI_ArrowArrayStream)>,
+    /// C function to get schema from the stream
+    pub get_schema:
+        Option<unsafe extern "C" fn(arg1: *mut Self, out: *mut FFI_ArrowSchema) -> c_int>,
+    /// C function to get next array from the stream
+    pub get_next: Option<unsafe extern "C" fn(arg1: *mut Self, out: *mut FFI_ArrowArray) -> c_int>,
+    /// C function to get the error from last operation on the stream
+    pub get_last_error: Option<unsafe extern "C" fn(arg1: *mut Self) -> *const c_char>,
+    /// C function to release the stream
+    pub release: Option<unsafe extern "C" fn(arg1: *mut Self)>,
+    /// Private data used by the stream
     pub private_data: *mut c_void,
 }
 
@@ -275,6 +276,7 @@ fn get_error_code(err: &ArrowError) -> i32 {
 }
 
 /// A `RecordBatchReader` which imports Arrays from `FFI_ArrowArrayStream`.
+///
 /// Struct used to fetch `RecordBatch` from the C Stream Interface.
 /// Its main responsibility is to expose `RecordBatchReader` functionality
 /// that requires [FFI_ArrowArrayStream].
diff --git a/arrow-array/src/types.rs b/arrow-array/src/types.rs
index 550d1aadf..b39c9c403 100644
--- a/arrow-array/src/types.rs
+++ b/arrow-array/src/types.rs
@@ -50,7 +50,9 @@ impl BooleanType {
     pub const DATA_TYPE: DataType = DataType::Boolean;
 }
 
-/// Trait for [primitive values], bridging the dynamic-typed nature of Arrow
+/// Trait for [primitive values].
+///
+/// This trait bridges the dynamic-typed nature of Arrow
 /// (via [`DataType`]) with the static-typed nature of rust types
 /// ([`ArrowNativeType`]) for all types that implement [`ArrowNativeType`].
 ///
diff --git a/arrow-avro/src/lib.rs b/arrow-avro/src/lib.rs
index f74edd7e9..d01d681b7 100644
--- a/arrow-avro/src/lib.rs
+++ b/arrow-avro/src/lib.rs
@@ -20,6 +20,7 @@
 //! [Apache Arrow]: https://arrow.apache.org
 //! [Apache Avro]: https://avro.apache.org/
 
+#![warn(missing_docs)]
 #![allow(unused)] // Temporary
 
 pub mod reader;
diff --git a/arrow-buffer/src/bigint/mod.rs b/arrow-buffer/src/bigint/mod.rs
index 0bd51e162..f5fab75dc 100644
--- a/arrow-buffer/src/bigint/mod.rs
+++ b/arrow-buffer/src/bigint/mod.rs
@@ -216,6 +216,7 @@ impl i256 {
         }
     }
 
+    /// Create an `i256` value from a 128-bit value.
     pub const fn from_i128(v: i128) -> Self {
         Self::from_parts(v as u128, v >> 127)
     }
diff --git a/arrow-buffer/src/buffer/immutable.rs b/arrow-buffer/src/buffer/immutable.rs
index 7cd355221..8d1a46583 100644
--- a/arrow-buffer/src/buffer/immutable.rs
+++ b/arrow-buffer/src/buffer/immutable.rs
@@ -26,7 +26,7 @@ use crate::BufferBuilder;
 use crate::{bytes::Bytes, native::ArrowNativeType};
 
 use super::ops::bitwise_unary_op_helper;
-use super::MutableBuffer;
+use super::{MutableBuffer, ScalarBuffer};
 
 /// Buffer represents a contiguous memory region that can be shared with other buffers and across
 /// thread boundaries.
@@ -203,7 +203,9 @@ impl Buffer {
     pub fn advance(&mut self, offset: usize) {
         assert!(
             offset <= self.length,
-            "the offset of the new Buffer cannot exceed the existing length"
+            "the offset of the new Buffer cannot exceed the existing length: offset={} length={}",
+            offset,
+            self.length
         );
         self.length -= offset;
         // Safety:
@@ -221,7 +223,8 @@ impl Buffer {
     pub fn slice_with_length(&self, offset: usize, length: usize) -> Self {
         assert!(
             offset.saturating_add(length) <= self.length,
-            "the offset of the new Buffer cannot exceed the existing length"
+            "the offset of the new Buffer cannot exceed the existing length: slice offset={offset} length={length} selflen={}",
+            self.length
         );
         // Safety:
         // offset + length <= self.length
@@ -388,6 +391,12 @@ impl<T: ArrowNativeType> From<Vec<T>> for Buffer {
     }
 }
 
+impl<T: ArrowNativeType> From<ScalarBuffer<T>> for Buffer {
+    fn from(value: ScalarBuffer<T>) -> Self {
+        value.into_inner()
+    }
+}
+
 /// Creating a `Buffer` instance by storing the boolean values into the buffer
 impl FromIterator<bool> for Buffer {
     fn from_iter<I>(iter: I) -> Self
diff --git a/arrow-buffer/src/builder/null.rs b/arrow-buffer/src/builder/null.rs
index a1cea6ef2..298b479e8 100644
--- a/arrow-buffer/src/builder/null.rs
+++ b/arrow-buffer/src/builder/null.rs
@@ -18,6 +18,7 @@
 use crate::{BooleanBufferBuilder, MutableBuffer, NullBuffer};
 
 /// Builder for creating the null bit buffer.
+///
 /// This builder only materializes the buffer when we append `false`.
 /// If you only append `true`s to the builder, what you get will be
 /// `None` when calling [`finish`](#method.finish).
@@ -158,6 +159,7 @@ impl NullBufferBuilder {
         }
     }
 
+    /// Return a mutable reference to the inner bitmap slice.
     pub fn as_slice_mut(&mut self) -> Option<&mut [u8]> {
         self.bitmap_builder.as_mut().map(|b| b.as_slice_mut())
     }
@@ -172,14 +174,12 @@ impl NullBufferBuilder {
 }
 
 impl NullBufferBuilder {
+    /// Return the number of bits in the buffer.
     pub fn len(&self) -> usize {
-        if let Some(b) = &self.bitmap_builder {
-            b.len()
-        } else {
-            self.len
-        }
+        self.bitmap_builder.as_ref().map_or(self.len, |b| b.len())
     }
 
+    /// Check if the builder is empty.
     pub fn is_empty(&self) -> bool {
         self.len() == 0
     }
diff --git a/arrow-buffer/src/builder/offset.rs b/arrow-buffer/src/builder/offset.rs
index 6a236d2a3..1ef0e3170 100644
--- a/arrow-buffer/src/builder/offset.rs
+++ b/arrow-buffer/src/builder/offset.rs
@@ -19,13 +19,13 @@ use std::ops::Deref;
 
 use crate::{ArrowNativeType, OffsetBuffer};
 
+/// Builder of [`OffsetBuffer`]
 #[derive(Debug)]
 pub struct OffsetBufferBuilder<O: ArrowNativeType> {
     offsets: Vec<O>,
     last_offset: usize,
 }
 
-/// Builder of [`OffsetBuffer`]
 impl<O: ArrowNativeType> OffsetBufferBuilder<O> {
     /// Create a new builder with space for `capacity + 1` offsets
     pub fn new(capacity: usize) -> Self {
diff --git a/arrow-buffer/src/interval.rs b/arrow-buffer/src/interval.rs
index 8f3342131..fa87fec6e 100644
--- a/arrow-buffer/src/interval.rs
+++ b/arrow-buffer/src/interval.rs
@@ -68,8 +68,11 @@ use std::ops::Neg;
 #[derive(Debug, Default, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
 #[repr(C)]
 pub struct IntervalMonthDayNano {
+    /// Number of months
     pub months: i32,
+    /// Number of days
     pub days: i32,
+    /// Number of nanoseconds
     pub nanoseconds: i64,
 }
 
@@ -345,7 +348,9 @@ derive_arith!(
 #[derive(Debug, Default, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
 #[repr(C)]
 pub struct IntervalDayTime {
+    /// Number of days
     pub days: i32,
+    /// Number of milliseconds
     pub milliseconds: i32,
 }
 
diff --git a/arrow-buffer/src/lib.rs b/arrow-buffer/src/lib.rs
index a7bf93ed0..34e432208 100644
--- a/arrow-buffer/src/lib.rs
+++ b/arrow-buffer/src/lib.rs
@@ -19,6 +19,7 @@
 
 // used by [`buffer::mutable::dangling_ptr`]
 #![cfg_attr(miri, feature(strict_provenance))]
+#![warn(missing_docs)]
 
 pub mod alloc;
 pub mod buffer;
diff --git a/arrow-buffer/src/util/bit_chunk_iterator.rs b/arrow-buffer/src/util/bit_chunk_iterator.rs
index 440450908..54995314c 100644
--- a/arrow-buffer/src/util/bit_chunk_iterator.rs
+++ b/arrow-buffer/src/util/bit_chunk_iterator.rs
@@ -131,26 +131,32 @@ impl<'a> UnalignedBitChunk<'a> {
         }
     }
 
+    /// Returns the number of leading padding bits
     pub fn lead_padding(&self) -> usize {
         self.lead_padding
     }
 
+    /// Returns the number of trailing padding bits
     pub fn trailing_padding(&self) -> usize {
         self.trailing_padding
     }
 
+    /// Returns the prefix, if any
     pub fn prefix(&self) -> Option<u64> {
         self.prefix
     }
 
+    /// Returns the suffix, if any
     pub fn suffix(&self) -> Option<u64> {
         self.suffix
     }
 
+    /// Returns reference to the chunks
     pub fn chunks(&self) -> &'a [u64] {
         self.chunks
     }
 
+    /// Returns an iterator over the chunks
     pub fn iter(&self) -> UnalignedBitChunkIterator<'a> {
         self.prefix
             .into_iter()
@@ -164,6 +170,7 @@ impl<'a> UnalignedBitChunk<'a> {
     }
 }
 
+/// Iterator over an [`UnalignedBitChunk`]
 pub type UnalignedBitChunkIterator<'a> = std::iter::Chain<
     std::iter::Chain<std::option::IntoIter<u64>, std::iter::Cloned<std::slice::Iter<'a, u64>>>,
     std::option::IntoIter<u64>,
@@ -212,6 +219,7 @@ pub struct BitChunks<'a> {
 }
 
 impl<'a> BitChunks<'a> {
+    /// Create a new [`BitChunks`] from a byte array, and an offset and length in bits
     pub fn new(buffer: &'a [u8], offset: usize, len: usize) -> Self {
         assert!(ceil(offset + len, 8) <= buffer.len() * 8);
 
@@ -232,6 +240,7 @@ impl<'a> BitChunks<'a> {
     }
 }
 
+/// Iterator over chunks of 64 bits represented as an u64
 #[derive(Debug)]
 pub struct BitChunkIterator<'a> {
     buffer: &'a [u8],
diff --git a/arrow-buffer/src/util/bit_mask.rs b/arrow-buffer/src/util/bit_mask.rs
index 8f81cb7d0..d4c2fa474 100644
--- a/arrow-buffer/src/util/bit_mask.rs
+++ b/arrow-buffer/src/util/bit_mask.rs
@@ -17,12 +17,14 @@
 
 //! Utils for working with packed bit masks
 
-use crate::bit_chunk_iterator::BitChunks;
-use crate::bit_util::{ceil, get_bit, set_bit};
+use crate::bit_util::ceil;
 
-/// Sets all bits on `write_data` in the range `[offset_write..offset_write+len]` to be equal to the
-/// bits in `data` in the range `[offset_read..offset_read+len]`
+/// Util function to set bits in a slice of bytes.
+///
+/// This will sets all bits on `write_data` in the range `[offset_write..offset_write+len]`
+/// to be equal to the bits in `data` in the range `[offset_read..offset_read+len]`
 /// returns the number of `0` bits `data[offset_read..offset_read+len]`
+/// `offset_write`, `offset_read`, and `len` are in terms of bits
 pub fn set_bits(
     write_data: &mut [u8],
     data: &[u8],
@@ -30,159 +32,401 @@ pub fn set_bits(
     offset_read: usize,
     len: usize,
 ) -> usize {
+    assert!(offset_write + len <= write_data.len() * 8);
+    assert!(offset_read + len <= data.len() * 8);
     let mut null_count = 0;
-
-    let mut bits_to_align = offset_write % 8;
-    if bits_to_align > 0 {
-        bits_to_align = std::cmp::min(len, 8 - bits_to_align);
+    let mut acc = 0;
+    while len > acc {
+        // SAFETY: the arguments to `set_upto_64bits` are within the valid range because
+        // (offset_write + acc) + (len - acc) == offset_write + len <= write_data.len() * 8
+        // (offset_read + acc) + (len - acc) == offset_read + len <= data.len() * 8
+        let (n, len_set) = unsafe {
+            set_upto_64bits(
+                write_data,
+                data,
+                offset_write + acc,
+                offset_read + acc,
+                len - acc,
+            )
+        };
+        null_count += n;
+        acc += len_set;
     }
-    let mut write_byte_index = ceil(offset_write + bits_to_align, 8);
-
-    // Set full bytes provided by bit chunk iterator (which iterates in 64 bits at a time)
-    let chunks = BitChunks::new(data, offset_read + bits_to_align, len - bits_to_align);
-    chunks.iter().for_each(|chunk| {
-        null_count += chunk.count_zeros();
-        write_data[write_byte_index..write_byte_index + 8].copy_from_slice(&chunk.to_le_bytes());
-        write_byte_index += 8;
-    });
-
-    // Set individual bits both to align write_data to a byte offset and the remainder bits not covered by the bit chunk iterator
-    let remainder_offset = len - chunks.remainder_len();
-    (0..bits_to_align)
-        .chain(remainder_offset..len)
-        .for_each(|i| {
-            if get_bit(data, offset_read + i) {
-                set_bit(write_data, offset_write + i);
+
+    null_count
+}
+
+/// Similar to `set_bits` but sets only upto 64 bits, actual number of bits set may vary.
+/// Returns a pair of the number of `0` bits and the number of bits set
+///
+/// # Safety
+/// The caller must ensure all arguments are within the valid range.
+#[inline]
+unsafe fn set_upto_64bits(
+    write_data: &mut [u8],
+    data: &[u8],
+    offset_write: usize,
+    offset_read: usize,
+    len: usize,
+) -> (usize, usize) {
+    let read_byte = offset_read / 8;
+    let read_shift = offset_read % 8;
+    let write_byte = offset_write / 8;
+    let write_shift = offset_write % 8;
+
+    if len >= 64 {
+        let chunk = unsafe { (data.as_ptr().add(read_byte) as *const u64).read_unaligned() };
+        if read_shift == 0 {
+            if write_shift == 0 {
+                // no shifting necessary
+                let len = 64;
+                let null_count = chunk.count_zeros() as usize;
+                unsafe { write_u64_bytes(write_data, write_byte, chunk) };
+                (null_count, len)
             } else {
-                null_count += 1;
+                // only write shifting necessary
+                let len = 64 - write_shift;
+                let chunk = chunk << write_shift;
+                let null_count = len - chunk.count_ones() as usize;
+                unsafe { or_write_u64_bytes(write_data, write_byte, chunk) };
+                (null_count, len)
             }
-        });
+        } else if write_shift == 0 {
+            // only read shifting necessary
+            let len = 64 - 8; // 56 bits so the next set_upto_64bits call will see write_shift == 0
+            let chunk = (chunk >> read_shift) & 0x00FFFFFFFFFFFFFF; // 56 bits mask
+            let null_count = len - chunk.count_ones() as usize;
+            unsafe { write_u64_bytes(write_data, write_byte, chunk) };
+            (null_count, len)
+        } else {
+            let len = 64 - std::cmp::max(read_shift, write_shift);
+            let chunk = (chunk >> read_shift) << write_shift;
+            let null_count = len - chunk.count_ones() as usize;
+            unsafe { or_write_u64_bytes(write_data, write_byte, chunk) };
+            (null_count, len)
+        }
+    } else if len == 1 {
+        let byte_chunk = (unsafe { data.get_unchecked(read_byte) } >> read_shift) & 1;
+        unsafe { *write_data.get_unchecked_mut(write_byte) |= byte_chunk << write_shift };
+        ((byte_chunk ^ 1) as usize, 1)
+    } else {
+        let len = std::cmp::min(len, 64 - std::cmp::max(read_shift, write_shift));
+        let bytes = ceil(len + read_shift, 8);
+        // SAFETY: the args of `read_bytes_to_u64` are valid as read_byte + bytes <= data.len()
+        let chunk = unsafe { read_bytes_to_u64(data, read_byte, bytes) };
+        let mask = u64::MAX >> (64 - len);
+        let chunk = (chunk >> read_shift) & mask; // masking to read `len` bits only
+        let chunk = chunk << write_shift; // shifting back to align with `write_data`
+        let null_count = len - chunk.count_ones() as usize;
+        let bytes = ceil(len + write_shift, 8);
+        for (i, c) in chunk.to_le_bytes().iter().enumerate().take(bytes) {
+            unsafe { *write_data.get_unchecked_mut(write_byte + i) |= c };
+        }
+        (null_count, len)
+    }
+}
+
+/// # Safety
+/// The caller must ensure all arguments are within the valid range.
+#[inline]
+unsafe fn read_bytes_to_u64(data: &[u8], offset: usize, count: usize) -> u64 {
+    debug_assert!(count <= 8);
+    let mut tmp = std::mem::MaybeUninit::<u64>::new(0);
+    let src = data.as_ptr().add(offset);
+    unsafe {
+        std::ptr::copy_nonoverlapping(src, tmp.as_mut_ptr() as *mut u8, count);
+        tmp.assume_init()
+    }
+}
 
-    null_count as usize
+/// # Safety
+/// The caller must ensure `data` has `offset..(offset + 8)` range
+#[inline]
+unsafe fn write_u64_bytes(data: &mut [u8], offset: usize, chunk: u64) {
+    let ptr = data.as_mut_ptr().add(offset) as *mut u64;
+    ptr.write_unaligned(chunk);
+}
+
+/// Similar to `write_u64_bytes`, but this method ORs the offset addressed `data` and `chunk`
+/// instead of overwriting
+///
+/// # Safety
+/// The caller must ensure `data` has `offset..(offset + 8)` range
+#[inline]
+unsafe fn or_write_u64_bytes(data: &mut [u8], offset: usize, chunk: u64) {
+    let ptr = data.as_mut_ptr().add(offset);
+    let chunk = chunk | (*ptr) as u64;
+    (ptr as *mut u64).write_unaligned(chunk);
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bit_util::{get_bit, set_bit, unset_bit};
+    use rand::prelude::StdRng;
+    use rand::{Fill, Rng, SeedableRng};
+    use std::fmt::Display;
 
     #[test]
     fn test_set_bits_aligned() {
-        let mut destination: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
-        let source: &[u8] = &[
-            0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
-            0b10100101,
-        ];
-
-        let destination_offset = 8;
-        let source_offset = 0;
-
-        let len = 64;
+        SetBitsTest {
+            write_data: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            data: vec![
+                0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
+                0b10100101,
+            ],
+            offset_write: 8,
+            offset_read: 0,
+            len: 64,
+            expected_data: vec![
+                0, 0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011,
+                0b11100111, 0b10100101, 0,
+            ],
+            expected_null_count: 24,
+        }
+        .verify();
+    }
 
-        let expected_data: &[u8] = &[
-            0, 0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
-            0b10100101, 0,
-        ];
-        let expected_null_count = 24;
-        let result = set_bits(
-            destination.as_mut_slice(),
-            source,
-            destination_offset,
-            source_offset,
-            len,
-        );
+    #[test]
+    fn test_set_bits_unaligned_destination_start() {
+        SetBitsTest {
+            write_data: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            data: vec![
+                0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
+                0b10100101,
+            ],
+            offset_write: 3,
+            offset_read: 0,
+            len: 64,
+            expected_data: vec![
+                0b00111000, 0b00101111, 0b11001101, 0b11011100, 0b01011110, 0b00011111, 0b00111110,
+                0b00101111, 0b00000101, 0b00000000,
+            ],
+            expected_null_count: 24,
+        }
+        .verify();
+    }
 
-        assert_eq!(destination, expected_data);
-        assert_eq!(result, expected_null_count);
+    #[test]
+    fn test_set_bits_unaligned_destination_end() {
+        SetBitsTest {
+            write_data: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            data: vec![
+                0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
+                0b10100101,
+            ],
+            offset_write: 8,
+            offset_read: 0,
+            len: 62,
+            expected_data: vec![
+                0, 0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011,
+                0b11100111, 0b00100101, 0,
+            ],
+            expected_null_count: 23,
+        }
+        .verify();
     }
 
     #[test]
-    fn test_set_bits_unaligned_destination_start() {
-        let mut destination: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
-        let source: &[u8] = &[
-            0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
-            0b10100101,
-        ];
+    fn test_set_bits_unaligned() {
+        SetBitsTest {
+            write_data: vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            data: vec![
+                0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
+                0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111, 0b10100101,
+                0b10011001, 0b11011011, 0b11101011, 0b11000011,
+            ],
+            offset_write: 3,
+            offset_read: 5,
+            len: 95,
+            expected_data: vec![
+                0b01111000, 0b01101001, 0b11100110, 0b11110110, 0b11111010, 0b11110000, 0b01111001,
+                0b01101001, 0b11100110, 0b11110110, 0b11111010, 0b11110000, 0b00000001,
+            ],
+            expected_null_count: 35,
+        }
+        .verify();
+    }
 
-        let destination_offset = 3;
-        let source_offset = 0;
+    #[test]
+    fn set_bits_fuzz() {
+        let mut rng = StdRng::seed_from_u64(42);
+        let mut data = SetBitsTest::new();
+        for _ in 0..100 {
+            data.regen(&mut rng);
+            data.verify();
+        }
+    }
 
-        let len = 64;
+    #[derive(Debug, Default)]
+    struct SetBitsTest {
+        /// target write data
+        write_data: Vec<u8>,
+        /// source data
+        data: Vec<u8>,
+        offset_write: usize,
+        offset_read: usize,
+        len: usize,
+        /// the expected contents of write_data after the test
+        expected_data: Vec<u8>,
+        /// the expected number of nulls copied at the end of the test
+        expected_null_count: usize,
+    }
 
-        let expected_data: &[u8] = &[
-            0b00111000, 0b00101111, 0b11001101, 0b11011100, 0b01011110, 0b00011111, 0b00111110,
-            0b00101111, 0b00000101, 0b00000000,
-        ];
-        let expected_null_count = 24;
-        let result = set_bits(
-            destination.as_mut_slice(),
-            source,
-            destination_offset,
-            source_offset,
-            len,
-        );
+    /// prints a byte slice as a binary string like "01010101 10101010"
+    struct BinaryFormatter<'a>(&'a [u8]);
+    impl<'a> Display for BinaryFormatter<'a> {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            for byte in self.0 {
+                write!(f, "{:08b} ", byte)?;
+            }
+            write!(f, " ")?;
+            Ok(())
+        }
+    }
 
-        assert_eq!(destination, expected_data);
-        assert_eq!(result, expected_null_count);
+    impl Display for SetBitsTest {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            writeln!(f, "SetBitsTest {{")?;
+            writeln!(f, "  write_data:    {}", BinaryFormatter(&self.write_data))?;
+            writeln!(f, "  data:          {}", BinaryFormatter(&self.data))?;
+            writeln!(
+                f,
+                "  expected_data: {}",
+                BinaryFormatter(&self.expected_data)
+            )?;
+            writeln!(f, "  offset_write: {}", self.offset_write)?;
+            writeln!(f, "  offset_read: {}", self.offset_read)?;
+            writeln!(f, "  len: {}", self.len)?;
+            writeln!(f, "  expected_null_count: {}", self.expected_null_count)?;
+            writeln!(f, "}}")
+        }
     }
 
-    #[test]
-    fn test_set_bits_unaligned_destination_end() {
-        let mut destination: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
-        let source: &[u8] = &[
-            0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
-            0b10100101,
-        ];
+    impl SetBitsTest {
+        /// create a new instance of FuzzData
+        fn new() -> Self {
+            Self::default()
+        }
 
-        let destination_offset = 8;
-        let source_offset = 0;
+        /// Update this instance's fields with randomly selected values and expected data
+        fn regen(&mut self, rng: &mut StdRng) {
+            //  (read) data
+            // ------------------+-----------------+-------
+            // .. offset_read .. | data            | ...
+            // ------------------+-----------------+-------
 
-        let len = 62;
+            // Write data
+            // -------------------+-----------------+-------
+            // .. offset_write .. | (data to write) | ...
+            // -------------------+-----------------+-------
 
-        let expected_data: &[u8] = &[
-            0, 0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
-            0b00100101, 0,
-        ];
-        let expected_null_count = 23;
-        let result = set_bits(
-            destination.as_mut_slice(),
-            source,
-            destination_offset,
-            source_offset,
-            len,
-        );
+            // length of data to copy
+            let len = rng.gen_range(0..=200);
 
-        assert_eq!(destination, expected_data);
-        assert_eq!(result, expected_null_count);
-    }
+            // randomly pick where we will write to
+            let offset_write_bits = rng.gen_range(0..=200);
+            let offset_write_bytes = if offset_write_bits % 8 == 0 {
+                offset_write_bits / 8
+            } else {
+                (offset_write_bits / 8) + 1
+            };
+            let extra_write_data_bytes = rng.gen_range(0..=5); // ensure 0 shows up often
 
-    #[test]
-    fn test_set_bits_unaligned() {
-        let mut destination: Vec<u8> = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
-        let source: &[u8] = &[
-            0b11100111, 0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111,
-            0b10100101, 0b10011001, 0b11011011, 0b11101011, 0b11000011, 0b11100111, 0b10100101,
-            0b10011001, 0b11011011, 0b11101011, 0b11000011,
-        ];
+            // randomly decide where we will read from
+            let extra_read_data_bytes = rng.gen_range(0..=5); // make sure 0 shows up often
+            let offset_read_bits = rng.gen_range(0..=200);
+            let offset_read_bytes = if offset_read_bits % 8 != 0 {
+                (offset_read_bits / 8) + 1
+            } else {
+                offset_read_bits / 8
+            };
+
+            // create space for writing
+            self.write_data.clear();
+            self.write_data
+                .resize(offset_write_bytes + len + extra_write_data_bytes, 0);
+
+            // interestingly set_bits seems to assume the output is already zeroed
+            // the fuzz tests fail when this is uncommented
+            //self.write_data.try_fill(rng).unwrap();
+            self.offset_write = offset_write_bits;
 
-        let destination_offset = 3;
-        let source_offset = 5;
+            // make source data
+            self.data
+                .resize(offset_read_bytes + len + extra_read_data_bytes, 0);
+            // fill source data with random bytes
+            self.data.try_fill(rng).unwrap();
+            self.offset_read = offset_read_bits;
 
-        let len = 95;
+            self.len = len;
 
-        let expected_data: &[u8] = &[
-            0b01111000, 0b01101001, 0b11100110, 0b11110110, 0b11111010, 0b11110000, 0b01111001,
-            0b01101001, 0b11100110, 0b11110110, 0b11111010, 0b11110000, 0b00000001,
+            // generated expectated output (not efficient)
+            self.expected_data.resize(self.write_data.len(), 0);
+            self.expected_data.copy_from_slice(&self.write_data);
+
+            self.expected_null_count = 0;
+            for i in 0..self.len {
+                let bit = get_bit(&self.data, self.offset_read + i);
+                if bit {
+                    set_bit(&mut self.expected_data, self.offset_write + i);
+                } else {
+                    unset_bit(&mut self.expected_data, self.offset_write + i);
+                    self.expected_null_count += 1;
+                }
+            }
+        }
+
+        /// call set_bits with the given parameters and compare with the expected output
+        fn verify(&self) {
+            // call set_bits and compare
+            let mut actual = self.write_data.to_vec();
+            let null_count = set_bits(
+                &mut actual,
+                &self.data,
+                self.offset_write,
+                self.offset_read,
+                self.len,
+            );
+
+            assert_eq!(actual, self.expected_data, "self: {}", self);
+            assert_eq!(null_count, self.expected_null_count, "self: {}", self);
+        }
+    }
+
+    #[test]
+    fn test_set_upto_64bits() {
+        // len >= 64
+        let write_data: &mut [u8] = &mut [0; 9];
+        let data: &[u8] = &[
+            0b00000001, 0b00000001, 0b00000001, 0b00000001, 0b00000001, 0b00000001, 0b00000001,
+            0b00000001, 0b00000001,
         ];
-        let expected_null_count = 35;
-        let result = set_bits(
-            destination.as_mut_slice(),
-            source,
-            destination_offset,
-            source_offset,
-            len,
+        let offset_write = 1;
+        let offset_read = 0;
+        let len = 65;
+        let (n, len_set) =
+            unsafe { set_upto_64bits(write_data, data, offset_write, offset_read, len) };
+        assert_eq!(n, 55);
+        assert_eq!(len_set, 63);
+        assert_eq!(
+            write_data,
+            &[
+                0b00000010, 0b00000010, 0b00000010, 0b00000010, 0b00000010, 0b00000010, 0b00000010,
+                0b00000010, 0b00000000
+            ]
         );
 
-        assert_eq!(destination, expected_data);
-        assert_eq!(result, expected_null_count);
+        // len = 1
+        let write_data: &mut [u8] = &mut [0b00000000];
+        let data: &[u8] = &[0b00000001];
+        let offset_write = 1;
+        let offset_read = 0;
+        let len = 1;
+        let (n, len_set) =
+            unsafe { set_upto_64bits(write_data, data, offset_write, offset_read, len) };
+        assert_eq!(n, 0);
+        assert_eq!(len_set, 1);
+        assert_eq!(write_data, &[0b00000010]);
     }
 }
diff --git a/arrow-cast/Cargo.toml b/arrow-cast/Cargo.toml
index 86f1a431a..4046f5226 100644
--- a/arrow-cast/Cargo.toml
+++ b/arrow-cast/Cargo.toml
@@ -49,7 +49,7 @@ arrow-select = { workspace = true }
 chrono = { workspace = true }
 half = { version = "2.1", default-features = false }
 num = { version = "0.4", default-features = false, features = ["std"] }
-lexical-core = { version = "^0.8", default-features = false, features = ["write-integers", "write-floats", "parse-integers", "parse-floats"] }
+lexical-core = { version = "1.0", default-features = false, features = ["write-integers", "write-floats", "parse-integers", "parse-floats"] }
 atoi = "2.0.0"
 comfy-table = { version = "7.0", optional = true, default-features = false }
 base64 = "0.22"
diff --git a/arrow-cast/src/cast/dictionary.rs b/arrow-cast/src/cast/dictionary.rs
index daaddc491..fc4d99430 100644
--- a/arrow-cast/src/cast/dictionary.rs
+++ b/arrow-cast/src/cast/dictionary.rs
@@ -202,11 +202,41 @@ pub(crate) fn cast_to_dictionary<K: ArrowDictionaryKeyType>(
         UInt16 => pack_numeric_to_dictionary::<K, UInt16Type>(array, dict_value_type, cast_options),
         UInt32 => pack_numeric_to_dictionary::<K, UInt32Type>(array, dict_value_type, cast_options),
         UInt64 => pack_numeric_to_dictionary::<K, UInt64Type>(array, dict_value_type, cast_options),
-        Decimal128(_, _) => {
-            pack_numeric_to_dictionary::<K, Decimal128Type>(array, dict_value_type, cast_options)
+        Decimal128(p, s) => {
+            let dict = pack_numeric_to_dictionary::<K, Decimal128Type>(
+                array,
+                dict_value_type,
+                cast_options,
+            )?;
+            let dict = dict
+                .as_dictionary::<K>()
+                .downcast_dict::<Decimal128Array>()
+                .unwrap();
+            let value = dict.values().clone();
+            // Set correct precision/scale
+            let value = value.with_precision_and_scale(p, s)?;
+            Ok(Arc::new(DictionaryArray::<K>::try_new(
+                dict.keys().clone(),
+                Arc::new(value),
+            )?))
         }
-        Decimal256(_, _) => {
-            pack_numeric_to_dictionary::<K, Decimal256Type>(array, dict_value_type, cast_options)
+        Decimal256(p, s) => {
+            let dict = pack_numeric_to_dictionary::<K, Decimal256Type>(
+                array,
+                dict_value_type,
+                cast_options,
+            )?;
+            let dict = dict
+                .as_dictionary::<K>()
+                .downcast_dict::<Decimal256Array>()
+                .unwrap();
+            let value = dict.values().clone();
+            // Set correct precision/scale
+            let value = value.with_precision_and_scale(p, s)?;
+            Ok(Arc::new(DictionaryArray::<K>::try_new(
+                dict.keys().clone(),
+                Arc::new(value),
+            )?))
         }
         Float16 => {
             pack_numeric_to_dictionary::<K, Float16Type>(array, dict_value_type, cast_options)
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index fe59a141c..e80d497c8 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -2650,6 +2650,38 @@ mod tests {
                    err.unwrap_err().to_string());
     }
 
+    #[test]
+    fn test_cast_decimal128_to_decimal128_dict() {
+        let p = 20;
+        let s = 3;
+        let input_type = DataType::Decimal128(p, s);
+        let output_type = DataType::Dictionary(
+            Box::new(DataType::Int32),
+            Box::new(DataType::Decimal128(p, s)),
+        );
+        assert!(can_cast_types(&input_type, &output_type));
+        let array = vec![Some(1123456), Some(2123456), Some(3123456), None];
+        let array = create_decimal_array(array, p, s).unwrap();
+        let cast_array = cast_with_options(&array, &output_type, &CastOptions::default()).unwrap();
+        assert_eq!(cast_array.data_type(), &output_type);
+    }
+
+    #[test]
+    fn test_cast_decimal256_to_decimal256_dict() {
+        let p = 20;
+        let s = 3;
+        let input_type = DataType::Decimal256(p, s);
+        let output_type = DataType::Dictionary(
+            Box::new(DataType::Int32),
+            Box::new(DataType::Decimal256(p, s)),
+        );
+        assert!(can_cast_types(&input_type, &output_type));
+        let array = vec![Some(1123456), Some(2123456), Some(3123456), None];
+        let array = create_decimal_array(array, p, s).unwrap();
+        let cast_array = cast_with_options(&array, &output_type, &CastOptions::default()).unwrap();
+        assert_eq!(cast_array.data_type(), &output_type);
+    }
+
     #[test]
     fn test_cast_decimal128_to_decimal128_overflow() {
         let input_type = DataType::Decimal128(38, 3);
diff --git a/arrow-cast/src/display.rs b/arrow-cast/src/display.rs
index 312e79739..df96816ea 100644
--- a/arrow-cast/src/display.rs
+++ b/arrow-cast/src/display.rs
@@ -81,6 +81,7 @@ impl<'a> Default for FormatOptions<'a> {
 }
 
 impl<'a> FormatOptions<'a> {
+    /// Creates a new set of format options
     pub const fn new() -> Self {
         Self {
             safe: true,
@@ -421,9 +422,7 @@ macro_rules! primitive_display {
             fn write(&self, idx: usize, f: &mut dyn Write) -> FormatResult {
                 let value = self.value(idx);
                 let mut buffer = [0u8; <$t as ArrowPrimitiveType>::Native::FORMATTED_SIZE];
-                // SAFETY:
-                // buffer is T::FORMATTED_SIZE
-                let b = unsafe { lexical_core::write_unchecked(value, &mut buffer) };
+                let b = lexical_core::write(value, &mut buffer);
                 // Lexical core produces valid UTF-8
                 let s = unsafe { std::str::from_utf8_unchecked(b) };
                 f.write_str(s)?;
diff --git a/arrow-cast/src/lib.rs b/arrow-cast/src/lib.rs
index 78acf2de5..6eac1be37 100644
--- a/arrow-cast/src/lib.rs
+++ b/arrow-cast/src/lib.rs
@@ -16,6 +16,8 @@
 // under the License.
 
 //! Functions for converting from one data type to another in [Apache Arrow](https://docs.rs/arrow)
+
+#![warn(missing_docs)]
 pub mod cast;
 pub use cast::*;
 pub mod display;
diff --git a/arrow-cast/src/parse.rs b/arrow-cast/src/parse.rs
index 20fa882e9..e332e5bba 100644
--- a/arrow-cast/src/parse.rs
+++ b/arrow-cast/src/parse.rs
@@ -432,8 +432,12 @@ fn string_to_time(s: &str) -> Option<NaiveTime> {
 /// assert_eq!(ts, 1609459200123456789);
 /// ```
 pub trait Parser: ArrowPrimitiveType {
+    /// Parse a string to the native type
     fn parse(string: &str) -> Option<Self::Native>;
 
+    /// Parse a string to the native type with a format string
+    ///
+    /// When not implemented, the format string is unused, and this method is equivalent to [parse](#tymethod.parse)
     fn parse_formatted(string: &str, _format: &str) -> Option<Self::Native> {
         Self::parse(string)
     }
@@ -966,6 +970,7 @@ pub fn parse_decimal<T: DecimalType>(
     })
 }
 
+/// Parse human-readable interval string to Arrow [IntervalYearMonthType]
 pub fn parse_interval_year_month(
     value: &str,
 ) -> Result<<IntervalYearMonthType as ArrowPrimitiveType>::Native, ArrowError> {
@@ -981,6 +986,7 @@ pub fn parse_interval_year_month(
     Ok(IntervalYearMonthType::make_value(0, months))
 }
 
+/// Parse human-readable interval string to Arrow [IntervalDayTimeType]
 pub fn parse_interval_day_time(
     value: &str,
 ) -> Result<<IntervalDayTimeType as ArrowPrimitiveType>::Native, ArrowError> {
@@ -994,6 +1000,7 @@ pub fn parse_interval_day_time(
     Ok(IntervalDayTimeType::make_value(days, millis))
 }
 
+/// Parse human-readable interval string to Arrow [IntervalMonthDayNanoType]
 pub fn parse_interval_month_day_nano_config(
     value: &str,
     config: IntervalParseConfig,
@@ -1005,6 +1012,7 @@ pub fn parse_interval_month_day_nano_config(
     Ok(IntervalMonthDayNanoType::make_value(months, days, nanos))
 }
 
+/// Parse human-readable interval string to Arrow [IntervalMonthDayNanoType]
 pub fn parse_interval_month_day_nano(
     value: &str,
 ) -> Result<<IntervalMonthDayNanoType as ArrowPrimitiveType>::Native, ArrowError> {
@@ -1018,14 +1026,18 @@ const NANOS_PER_HOUR: i64 = 60 * NANOS_PER_MINUTE;
 #[cfg(test)]
 const NANOS_PER_DAY: i64 = 24 * NANOS_PER_HOUR;
 
+/// Config to parse interval strings
+///
+/// Currently stores the `default_unit` to use if the string doesn't have one specified
 #[derive(Debug, Clone)]
 pub struct IntervalParseConfig {
     /// The default unit to use if none is specified
-    /// e.g. `INTERVAL 1` represents `INTERVAL 1 SECOND` when default_unit = IntervalType::Second
+    /// e.g. `INTERVAL 1` represents `INTERVAL 1 SECOND` when default_unit = [IntervalUnit::Second]
     default_unit: IntervalUnit,
 }
 
 impl IntervalParseConfig {
+    /// Create a new [IntervalParseConfig] with the given default unit
     pub fn new(default_unit: IntervalUnit) -> Self {
         Self { default_unit }
     }
@@ -1034,18 +1046,32 @@ impl IntervalParseConfig {
 #[rustfmt::skip]
 #[derive(Debug, Clone, Copy)]
 #[repr(u16)]
+/// Represents the units of an interval, with each variant
+/// corresponding to a bit in the interval's bitfield representation
 pub enum IntervalUnit {
+    /// A Century
     Century     = 0b_0000_0000_0001,
+    /// A Decade
     Decade      = 0b_0000_0000_0010,
+    /// A Year
     Year        = 0b_0000_0000_0100,
+    /// A Month
     Month       = 0b_0000_0000_1000,
+    /// A Week
     Week        = 0b_0000_0001_0000,
+    /// A Day
     Day         = 0b_0000_0010_0000,
+    /// An Hour
     Hour        = 0b_0000_0100_0000,
+    /// A Minute
     Minute      = 0b_0000_1000_0000,
+    /// A Second
     Second      = 0b_0001_0000_0000,
+    /// A Millisecond
     Millisecond = 0b_0010_0000_0000,
+    /// A Microsecond
     Microsecond = 0b_0100_0000_0000,
+    /// A Nanosecond
     Nanosecond  = 0b_1000_0000_0000,
 }
 
@@ -1093,6 +1119,7 @@ impl IntervalUnit {
     }
 }
 
+/// A tuple representing (months, days, nanoseconds) in an interval
 pub type MonthDayNano = (i32, i32, i64);
 
 /// Chosen based on the number of decimal digits in 1 week in nanoseconds
diff --git a/arrow-cast/src/pretty.rs b/arrow-cast/src/pretty.rs
index f41471e38..4a3cbda28 100644
--- a/arrow-cast/src/pretty.rs
+++ b/arrow-cast/src/pretty.rs
@@ -54,7 +54,8 @@ pub fn pretty_format_columns(
     pretty_format_columns_with_options(col_name, results, &options)
 }
 
-pub fn pretty_format_columns_with_options(
+/// Utility function to create a visual representation of columns with options
+fn pretty_format_columns_with_options(
     col_name: &str,
     results: &[ArrayRef],
     options: &FormatOptions,
diff --git a/arrow-csv/Cargo.toml b/arrow-csv/Cargo.toml
index d29c85c56..be213c936 100644
--- a/arrow-csv/Cargo.toml
+++ b/arrow-csv/Cargo.toml
@@ -43,7 +43,7 @@ chrono = { workspace = true }
 csv = { version = "1.1", default-features = false }
 csv-core = { version = "0.1" }
 lazy_static = { version = "1.4", default-features = false }
-lexical-core = { version = "^0.8", default-features = false }
+lexical-core = { version = "1.0", default-features = false }
 regex = { version = "1.7.0", default-features = false, features = ["std", "unicode", "perf"] }
 
 [dev-dependencies]
diff --git a/arrow-csv/src/lib.rs b/arrow-csv/src/lib.rs
index e6dc69935..28c0d6ebd 100644
--- a/arrow-csv/src/lib.rs
+++ b/arrow-csv/src/lib.rs
@@ -17,6 +17,8 @@
 
 //! Transfer data between the Arrow memory format and CSV (comma-separated values).
 
+#![warn(missing_docs)]
+
 pub mod reader;
 pub mod writer;
 
diff --git a/arrow-csv/src/reader/mod.rs b/arrow-csv/src/reader/mod.rs
index c5057599b..36f80ec90 100644
--- a/arrow-csv/src/reader/mod.rs
+++ b/arrow-csv/src/reader/mod.rs
@@ -236,31 +236,41 @@ pub struct Format {
 }
 
 impl Format {
+    /// Specify whether the CSV file has a header, defaults to `true`
+    ///
+    /// When `true`, the first row of the CSV file is treated as a header row
     pub fn with_header(mut self, has_header: bool) -> Self {
         self.header = has_header;
         self
     }
 
+    /// Specify a custom delimiter character, defaults to comma `','`
     pub fn with_delimiter(mut self, delimiter: u8) -> Self {
         self.delimiter = Some(delimiter);
         self
     }
 
+    /// Specify an escape character, defaults to `None`
     pub fn with_escape(mut self, escape: u8) -> Self {
         self.escape = Some(escape);
         self
     }
 
+    /// Specify a custom quote character, defaults to double quote `'"'`
     pub fn with_quote(mut self, quote: u8) -> Self {
         self.quote = Some(quote);
         self
     }
 
+    /// Specify a custom terminator character, defaults to CRLF
     pub fn with_terminator(mut self, terminator: u8) -> Self {
         self.terminator = Some(terminator);
         self
     }
 
+    /// Specify a comment character, defaults to `None`
+    ///
+    /// Lines starting with this character will be ignored
     pub fn with_comment(mut self, comment: u8) -> Self {
         self.comment = Some(comment);
         self
@@ -1100,7 +1110,7 @@ impl ReaderBuilder {
         self
     }
 
-    /// Overrides the [`Format`] of this [`ReaderBuilder]
+    /// Overrides the [Format] of this [ReaderBuilder]
     pub fn with_format(mut self, format: Format) -> Self {
         self.format = format;
         self
@@ -1112,21 +1122,25 @@ impl ReaderBuilder {
         self
     }
 
+    /// Set the given character as the CSV file's escape character
     pub fn with_escape(mut self, escape: u8) -> Self {
         self.format.escape = Some(escape);
         self
     }
 
+    /// Set the given character as the CSV file's quote character, by default it is double quote
     pub fn with_quote(mut self, quote: u8) -> Self {
         self.format.quote = Some(quote);
         self
     }
 
+    /// Provide a custom terminator character, defaults to CRLF
     pub fn with_terminator(mut self, terminator: u8) -> Self {
         self.format.terminator = Some(terminator);
         self
     }
 
+    /// Provide a comment character, lines starting with this character will be ignored
     pub fn with_comment(mut self, comment: u8) -> Self {
         self.format.comment = Some(comment);
         self
diff --git a/arrow-data/src/data.rs b/arrow-data/src/data.rs
index a14bc4873..33cbc897a 100644
--- a/arrow-data/src/data.rs
+++ b/arrow-data/src/data.rs
@@ -161,9 +161,11 @@ pub(crate) fn new_buffers(data_type: &DataType, capacity: usize) -> [MutableBuff
     }
 }
 
-/// A generic representation of Arrow array data which encapsulates common attributes and
-/// operations for Arrow array. Specific operations for different arrays types (e.g.,
-/// primitive, list, struct) are implemented in `Array`.
+/// A generic representation of Arrow array data which encapsulates common attributes
+/// and operations for Arrow array.
+///
+/// Specific operations for different arrays types (e.g., primitive, list, struct)
+/// are implemented in `Array`.
 ///
 /// # Memory Layout
 ///
diff --git a/arrow-data/src/ffi.rs b/arrow-data/src/ffi.rs
index 3345595fa..cd283d326 100644
--- a/arrow-data/src/ffi.rs
+++ b/arrow-data/src/ffi.rs
@@ -20,7 +20,7 @@
 use crate::bit_mask::set_bits;
 use crate::{layout, ArrayData};
 use arrow_buffer::buffer::NullBuffer;
-use arrow_buffer::{Buffer, MutableBuffer};
+use arrow_buffer::{Buffer, MutableBuffer, ScalarBuffer};
 use arrow_schema::DataType;
 use std::ffi::c_void;
 
@@ -121,7 +121,7 @@ impl FFI_ArrowArray {
     pub fn new(data: &ArrayData) -> Self {
         let data_layout = layout(data.data_type());
 
-        let buffers = if data_layout.can_contain_null_mask {
+        let mut buffers = if data_layout.can_contain_null_mask {
             // * insert the null buffer at the start
             // * make all others `Option<Buffer>`.
             std::iter::once(align_nulls(data.offset(), data.nulls()))
@@ -132,7 +132,7 @@ impl FFI_ArrowArray {
         };
 
         // `n_buffers` is the number of buffers by the spec.
-        let n_buffers = {
+        let mut n_buffers = {
             data_layout.buffers.len() + {
                 // If the layout has a null buffer by Arrow spec.
                 // Note that even the array doesn't have a null buffer because it has
@@ -141,10 +141,22 @@ impl FFI_ArrowArray {
             }
         } as i64;
 
+        if data_layout.variadic {
+            // Save the lengths of all variadic buffers into a new buffer.
+            // The first buffer is `views`, and the rest are variadic.
+            let mut data_buffers_lengths = Vec::new();
+            for buffer in data.buffers().iter().skip(1) {
+                data_buffers_lengths.push(buffer.len() as i64);
+                n_buffers += 1;
+            }
+
+            buffers.push(Some(ScalarBuffer::from(data_buffers_lengths).into_inner()));
+            n_buffers += 1;
+        }
+
         let buffers_ptr = buffers
             .iter()
             .flat_map(|maybe_buffer| match maybe_buffer {
-                // note that `raw_data` takes into account the buffer's offset
                 Some(b) => Some(b.as_ptr() as *const c_void),
                 // This is for null buffer. We only put a null pointer for
                 // null buffer if by spec it can contain null mask.
diff --git a/arrow-flight/gen/Cargo.toml b/arrow-flight/gen/Cargo.toml
index ae6000ae7..e4801cdaa 100644
--- a/arrow-flight/gen/Cargo.toml
+++ b/arrow-flight/gen/Cargo.toml
@@ -33,5 +33,5 @@ publish = false
 # Pin specific version of the tonic-build dependencies to avoid auto-generated
 # (and checked in) arrow.flight.protocol.rs from changing
 proc-macro2 = { version = "=1.0.86", default-features = false }
-prost-build = { version = "=0.13.1", default-features = false }
+prost-build = { version = "=0.13.2", default-features = false }
 tonic-build = { version = "=0.12.2", default-features = false, features = ["transport", "prost"] }
diff --git a/arrow-flight/src/arrow.flight.protocol.rs b/arrow-flight/src/arrow.flight.protocol.rs
index 670d4a649..f3ae7774a 100644
--- a/arrow-flight/src/arrow.flight.protocol.rs
+++ b/arrow-flight/src/arrow.flight.protocol.rs
@@ -3,7 +3,6 @@
 // This file is @generated by prost-build.
 ///
 /// The request that a client provides to a server on handshake.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct HandshakeRequest {
     ///
@@ -15,7 +14,6 @@ pub struct HandshakeRequest {
     #[prost(bytes = "bytes", tag = "2")]
     pub payload: ::prost::bytes::Bytes,
 }
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct HandshakeResponse {
     ///
@@ -29,7 +27,6 @@ pub struct HandshakeResponse {
 }
 ///
 /// A message for doing simple auth.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BasicAuth {
     #[prost(string, tag = "2")]
@@ -37,13 +34,11 @@ pub struct BasicAuth {
     #[prost(string, tag = "3")]
     pub password: ::prost::alloc::string::String,
 }
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct Empty {}
 ///
 /// Describes an available action, including both the name used for execution
 /// along with a short description of the purpose of the action.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionType {
     #[prost(string, tag = "1")]
@@ -54,7 +49,6 @@ pub struct ActionType {
 ///
 /// A service specific expression that can be used to return a limited set
 /// of available Arrow Flight streams.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Criteria {
     #[prost(bytes = "bytes", tag = "1")]
@@ -62,7 +56,6 @@ pub struct Criteria {
 }
 ///
 /// An opaque action specific for the service.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Action {
     #[prost(string, tag = "1")]
@@ -74,7 +67,6 @@ pub struct Action {
 /// The request of the CancelFlightInfo action.
 ///
 /// The request should be stored in Action.body.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CancelFlightInfoRequest {
     #[prost(message, optional, tag = "1")]
@@ -84,7 +76,6 @@ pub struct CancelFlightInfoRequest {
 /// The request of the RenewFlightEndpoint action.
 ///
 /// The request should be stored in Action.body.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct RenewFlightEndpointRequest {
     #[prost(message, optional, tag = "1")]
@@ -92,7 +83,6 @@ pub struct RenewFlightEndpointRequest {
 }
 ///
 /// An opaque result returned after executing an action.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Result {
     #[prost(bytes = "bytes", tag = "1")]
@@ -102,7 +92,6 @@ pub struct Result {
 /// The result of the CancelFlightInfo action.
 ///
 /// The result should be stored in Result.body.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CancelFlightInfoResult {
     #[prost(enumeration = "CancelStatus", tag = "1")]
@@ -110,7 +99,6 @@ pub struct CancelFlightInfoResult {
 }
 ///
 /// Wrap the result of a getSchema call
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SchemaResult {
     /// The schema of the dataset in its IPC form:
@@ -123,7 +111,6 @@ pub struct SchemaResult {
 ///
 /// The name or tag for a Flight. May be used as a way to retrieve or generate
 /// a flight or be used to expose a set of previously defined flights.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightDescriptor {
     #[prost(enumeration = "flight_descriptor::DescriptorType", tag = "1")]
@@ -193,7 +180,6 @@ pub mod flight_descriptor {
 ///
 /// The access coordinates for retrieval of a dataset. With a FlightInfo, a
 /// consumer is able to determine how to retrieve a dataset.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightInfo {
     /// The schema of the dataset in its IPC form:
@@ -253,7 +239,6 @@ pub struct FlightInfo {
 }
 ///
 /// The information to process a long-running query.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PollInfo {
     ///
@@ -294,7 +279,6 @@ pub struct PollInfo {
 }
 ///
 /// A particular stream or split associated with a flight.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightEndpoint {
     ///
@@ -338,7 +322,6 @@ pub struct FlightEndpoint {
 ///
 /// A location where a Flight service will accept retrieval of a particular
 /// stream given a ticket.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Location {
     #[prost(string, tag = "1")]
@@ -350,7 +333,6 @@ pub struct Location {
 ///
 /// Tickets are meant to be single use. It is an error/application-defined
 /// behavior to reuse a ticket.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Ticket {
     #[prost(bytes = "bytes", tag = "1")]
@@ -358,7 +340,6 @@ pub struct Ticket {
 }
 ///
 /// A batch of Arrow data as part of a stream of batches.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct FlightData {
     ///
@@ -384,7 +365,6 @@ pub struct FlightData {
 }
 /// *
 /// The response message associated with the submission of a DoPut.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct PutResult {
     #[prost(bytes = "bytes", tag = "1")]
diff --git a/arrow-flight/src/lib.rs b/arrow-flight/src/lib.rs
index 1180264e5..ff9e387da 100644
--- a/arrow-flight/src/lib.rs
+++ b/arrow-flight/src/lib.rs
@@ -50,8 +50,7 @@ use std::{fmt, ops::Deref};
 
 type ArrowResult<T> = std::result::Result<T, ArrowError>;
 
-#[allow(clippy::derive_partial_eq_without_eq)]
-
+#[allow(clippy::all)]
 mod gen {
     include!("arrow.flight.protocol.rs");
 }
diff --git a/arrow-flight/src/sql/arrow.flight.protocol.sql.rs b/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
index 3eeed6ff4..805759cde 100644
--- a/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
+++ b/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
@@ -19,7 +19,6 @@
 ///               int32_to_int32_list_map: map<key: int32, value: list<$data$: int32>>
 /// >
 /// where there is one row per requested piece of metadata information.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetSqlInfo {
     ///
@@ -100,7 +99,6 @@ pub struct CommandGetSqlInfo {
 ///                               is only relevant to be used by ODBC).
 /// >
 /// The returned data should be ordered by data_type and then by type_name.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CommandGetXdbcTypeInfo {
     ///
@@ -120,7 +118,6 @@ pub struct CommandGetXdbcTypeInfo {
 ///   catalog_name: utf8 not null
 /// >
 /// The returned data should be ordered by catalog_name.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CommandGetCatalogs {}
 ///
@@ -136,7 +133,6 @@ pub struct CommandGetCatalogs {}
 ///   db_schema_name: utf8 not null
 /// >
 /// The returned data should be ordered by catalog_name, then db_schema_name.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetDbSchemas {
     ///
@@ -181,7 +177,6 @@ pub struct CommandGetDbSchemas {
 ///   - ARROW:FLIGHT:SQL:IS_READ_ONLY      - "1" indicates if the column is read only, "0" otherwise.
 ///   - ARROW:FLIGHT:SQL:IS_SEARCHABLE     - "1" indicates if the column is searchable via WHERE clause, "0" otherwise.
 /// The returned data should be ordered by catalog_name, db_schema_name, table_name, then table_type, followed by table_schema if requested.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetTables {
     ///
@@ -231,7 +226,6 @@ pub struct CommandGetTables {
 ///   table_type: utf8 not null
 /// >
 /// The returned data should be ordered by table_type.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct CommandGetTableTypes {}
 ///
@@ -250,7 +244,6 @@ pub struct CommandGetTableTypes {}
 ///   key_sequence: int32 not null
 /// >
 /// The returned data should be ordered by catalog_name, db_schema_name, table_name, key_name, then key_sequence.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetPrimaryKeys {
     ///
@@ -294,7 +287,6 @@ pub struct CommandGetPrimaryKeys {
 /// >
 /// The returned data should be ordered by fk_catalog_name, fk_db_schema_name, fk_table_name, fk_key_name, then key_sequence.
 /// update_rule and delete_rule returns a byte that is equivalent to actions declared on UpdateDeleteRules enum.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetExportedKeys {
     ///
@@ -342,7 +334,6 @@ pub struct CommandGetExportedKeys {
 ///     - 2 = SET NULL
 ///     - 3 = NO ACTION
 ///     - 4 = SET DEFAULT
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetImportedKeys {
     ///
@@ -392,7 +383,6 @@ pub struct CommandGetImportedKeys {
 ///     - 2 = SET NULL
 ///     - 3 = NO ACTION
 ///     - 4 = SET DEFAULT
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandGetCrossReference {
     /// *
@@ -430,7 +420,6 @@ pub struct CommandGetCrossReference {
 }
 ///
 /// Request message for the "CreatePreparedStatement" action on a Flight SQL enabled backend.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCreatePreparedStatementRequest {
     /// The valid SQL string to create a prepared statement for.
@@ -443,7 +432,6 @@ pub struct ActionCreatePreparedStatementRequest {
 }
 ///
 /// An embedded message describing a Substrait plan to execute.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SubstraitPlan {
     /// The serialized substrait.Plan to create a prepared statement for.
@@ -460,7 +448,6 @@ pub struct SubstraitPlan {
 }
 ///
 /// Request message for the "CreatePreparedSubstraitPlan" action on a Flight SQL enabled backend.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCreatePreparedSubstraitPlanRequest {
     /// The serialized substrait.Plan to create a prepared statement for.
@@ -479,7 +466,6 @@ pub struct ActionCreatePreparedSubstraitPlanRequest {
 /// - Automatically, by a server timeout.
 ///
 /// The result should be wrapped in a google.protobuf.Any message.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCreatePreparedStatementResult {
     /// Opaque handle for the prepared statement on the server.
@@ -500,7 +486,6 @@ pub struct ActionCreatePreparedStatementResult {
 ///
 /// Request message for the "ClosePreparedStatement" action on a Flight SQL enabled backend.
 /// Closes server resources associated with the prepared statement handle.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionClosePreparedStatementRequest {
     /// Opaque handle for the prepared statement on the server.
@@ -510,7 +495,6 @@ pub struct ActionClosePreparedStatementRequest {
 ///
 /// Request message for the "BeginTransaction" action.
 /// Begins a transaction.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ActionBeginTransactionRequest {}
 ///
@@ -519,7 +503,6 @@ pub struct ActionBeginTransactionRequest {}
 ///
 /// Only supported if FLIGHT_SQL_TRANSACTION is
 /// FLIGHT_SQL_TRANSACTION_SUPPORT_SAVEPOINT.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionBeginSavepointRequest {
     /// The transaction to which a savepoint belongs.
@@ -537,7 +520,6 @@ pub struct ActionBeginSavepointRequest {
 /// automatically rolled back.
 ///
 /// The result should be wrapped in a google.protobuf.Any message.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionBeginTransactionResult {
     /// Opaque handle for the transaction on the server.
@@ -552,7 +534,6 @@ pub struct ActionBeginTransactionResult {
 /// out, then the savepoint is also invalidated.
 ///
 /// The result should be wrapped in a google.protobuf.Any message.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionBeginSavepointResult {
     /// Opaque handle for the savepoint on the server.
@@ -566,7 +547,6 @@ pub struct ActionBeginSavepointResult {
 ///
 /// If the action completes successfully, the transaction handle is
 /// invalidated, as are all associated savepoints.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionEndTransactionRequest {
     /// Opaque handle for the transaction on the server.
@@ -629,7 +609,6 @@ pub mod action_end_transaction_request {
 /// Releasing a savepoint invalidates that savepoint.  Rolling back to
 /// a savepoint does not invalidate the savepoint, but invalidates all
 /// savepoints created after the current savepoint.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionEndSavepointRequest {
     /// Opaque handle for the savepoint on the server.
@@ -699,7 +678,6 @@ pub mod action_end_savepoint_request {
 ///     - ARROW:FLIGHT:SQL:IS_READ_ONLY      - "1" indicates if the column is read only, "0" otherwise.
 ///     - ARROW:FLIGHT:SQL:IS_SEARCHABLE     - "1" indicates if the column is searchable via WHERE clause, "0" otherwise.
 ///   - GetFlightInfo: execute the query.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementQuery {
     /// The SQL syntax.
@@ -726,7 +704,6 @@ pub struct CommandStatementQuery {
 ///     - ARROW:FLIGHT:SQL:IS_SEARCHABLE     - "1" indicates if the column is searchable via WHERE clause, "0" otherwise.
 ///   - GetFlightInfo: execute the query.
 ///   - DoPut: execute the query.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementSubstraitPlan {
     /// A serialized substrait.Plan
@@ -739,7 +716,6 @@ pub struct CommandStatementSubstraitPlan {
 /// *
 /// Represents a ticket resulting from GetFlightInfo with a CommandStatementQuery.
 /// This should be used only once and treated as an opaque value, that is, clients should not attempt to parse this.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct TicketStatementQuery {
     /// Unique identifier for the instance of the statement to execute.
@@ -766,7 +742,6 @@ pub struct TicketStatementQuery {
 ///     for the parameters when determining the schema.
 ///   - DoPut: bind parameter values. All of the bound parameter sets will be executed as a single atomic execution.
 ///   - GetFlightInfo: execute the prepared statement instance.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandPreparedStatementQuery {
     /// Opaque handle for the prepared statement on the server.
@@ -776,7 +751,6 @@ pub struct CommandPreparedStatementQuery {
 ///
 /// Represents a SQL update query. Used in the command member of FlightDescriptor
 /// for the RPC call DoPut to cause the server to execute the included SQL update.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementUpdate {
     /// The SQL syntax.
@@ -790,7 +764,6 @@ pub struct CommandStatementUpdate {
 /// Represents a SQL update query. Used in the command member of FlightDescriptor
 /// for the RPC call DoPut to cause the server to execute the included
 /// prepared statement handle as an update.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandPreparedStatementUpdate {
     /// Opaque handle for the prepared statement on the server.
@@ -801,7 +774,6 @@ pub struct CommandPreparedStatementUpdate {
 /// Represents a bulk ingestion request. Used in the command member of FlightDescriptor
 /// for the the RPC call DoPut to cause the server load the contents of the stream's
 /// FlightData into the target destination.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandStatementIngest {
     /// The behavior for handling the table definition.
@@ -838,7 +810,6 @@ pub struct CommandStatementIngest {
 /// Nested message and enum types in `CommandStatementIngest`.
 pub mod command_statement_ingest {
     /// Options for table definition behavior
-    #[allow(clippy::derive_partial_eq_without_eq)]
     #[derive(Clone, Copy, PartialEq, ::prost::Message)]
     pub struct TableDefinitionOptions {
         #[prost(
@@ -949,7 +920,6 @@ pub mod command_statement_ingest {
 /// Returned from the RPC call DoPut when a CommandStatementUpdate,
 /// CommandPreparedStatementUpdate, or CommandStatementIngest was
 /// in the request, containing results from the update.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct DoPutUpdateResult {
     /// The number of records updated. A return value of -1 represents
@@ -962,7 +932,6 @@ pub struct DoPutUpdateResult {
 /// *Note on legacy behavior*: previous versions of the protocol did not return any result for
 /// this command, and that behavior should still be supported by clients. In that case, the client
 /// can continue as though the fields in this message were not provided or set to sensible default values.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct DoPutPreparedStatementResult {
     /// Represents a (potentially updated) opaque handle for the prepared statement on the server.
@@ -992,7 +961,6 @@ pub struct DoPutPreparedStatementResult {
 ///
 /// This command is deprecated since 13.0.0. Use the "CancelFlightInfo"
 /// action with DoAction instead.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct ActionCancelQueryRequest {
     /// The result of the GetFlightInfo RPC that initiated the query.
@@ -1009,7 +977,6 @@ pub struct ActionCancelQueryRequest {
 ///
 /// This command is deprecated since 13.0.0. Use the "CancelFlightInfo"
 /// action with DoAction instead.
-#[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, Copy, PartialEq, ::prost::Message)]
 pub struct ActionCancelQueryResult {
     #[prost(enumeration = "action_cancel_query_result::CancelResult", tag = "1")]
diff --git a/arrow-flight/src/sql/server.rs b/arrow-flight/src/sql/server.rs
index e348367a9..37b2885b5 100644
--- a/arrow-flight/src/sql/server.rs
+++ b/arrow-flight/src/sql/server.rs
@@ -979,6 +979,7 @@ fn arrow_error_to_status(err: arrow_schema::ArrowError) -> Status {
 
 /// A wrapper around [`Streaming<FlightData>`] that allows "peeking" at the
 /// message at the front of the stream without consuming it.
+///
 /// This is needed because sometimes the first message in the stream will contain
 /// a [`FlightDescriptor`] in addition to potentially any data, and the dispatch logic
 /// must inspect this information.
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index 2b1d09dc9..3e07c95af 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -149,7 +149,7 @@ fn create_array(
             // still work
             for struct_field in struct_fields {
                 let child = create_array(reader, struct_field, variadic_counts, require_alignment)?;
-                struct_arrays.push((struct_field.clone(), child));
+                struct_arrays.push(child);
             }
             let null_count = struct_node.null_count() as usize;
             let struct_array = if struct_arrays.is_empty() {
@@ -162,9 +162,11 @@ fn create_array(
                 )
             } else if null_count > 0 {
                 // create struct array from fields, arrays and null data
-                StructArray::from((struct_arrays, null_buffer))
+                let len = struct_node.length() as usize;
+                let nulls = BooleanBuffer::new(null_buffer, 0, len).into();
+                StructArray::try_new(struct_fields.clone(), struct_arrays, Some(nulls))?
             } else {
-                StructArray::from(struct_arrays)
+                StructArray::try_new(struct_fields.clone(), struct_arrays, None)?
             };
             Ok(Arc::new(struct_array))
         }
@@ -2235,4 +2237,50 @@ mod tests {
 
         assert_eq!(batch, roundtrip_batch);
     }
+
+    #[test]
+    fn test_invalid_struct_array_ipc_read_errors() {
+        let a_field = Field::new("a", DataType::Int32, false);
+        let b_field = Field::new("b", DataType::Int32, false);
+
+        let schema = Arc::new(Schema::new(vec![Field::new_struct(
+            "s",
+            vec![a_field.clone(), b_field.clone()],
+            false,
+        )]));
+
+        let a_array_data = ArrayData::builder(a_field.data_type().clone())
+            .len(4)
+            .add_buffer(Buffer::from_slice_ref([1, 2, 3, 4]))
+            .build()
+            .unwrap();
+        let b_array_data = ArrayData::builder(b_field.data_type().clone())
+            .len(3)
+            .add_buffer(Buffer::from_slice_ref([5, 6, 7]))
+            .build()
+            .unwrap();
+
+        let struct_data_type = schema.field(0).data_type();
+
+        let invalid_struct_arr = unsafe {
+            make_array(
+                ArrayData::builder(struct_data_type.clone())
+                    .len(4)
+                    .add_child_data(a_array_data)
+                    .add_child_data(b_array_data)
+                    .build_unchecked(),
+            )
+        };
+
+        let batch = RecordBatch::try_new(schema.clone(), vec![invalid_struct_arr]).unwrap();
+
+        let mut buf = Vec::new();
+        let mut writer = crate::writer::FileWriter::try_new(&mut buf, schema.as_ref()).unwrap();
+        writer.write(&batch).unwrap();
+        writer.finish().unwrap();
+
+        let mut reader = FileReader::try_new(std::io::Cursor::new(buf), None).unwrap();
+        let err = reader.next().unwrap().unwrap_err();
+        assert!(matches!(err, ArrowError::InvalidArgumentError(_)));
+    }
 }
diff --git a/arrow-ipc/src/reader/stream.rs b/arrow-ipc/src/reader/stream.rs
index 64191a22b..de5f5bdd6 100644
--- a/arrow-ipc/src/reader/stream.rs
+++ b/arrow-ipc/src/reader/stream.rs
@@ -275,8 +275,10 @@ impl StreamDecoder {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::writer::StreamWriter;
-    use arrow_array::{Int32Array, Int64Array, RecordBatch};
+    use crate::writer::{IpcWriteOptions, StreamWriter};
+    use arrow_array::{
+        types::Int32Type, DictionaryArray, Int32Array, Int64Array, RecordBatch, RunArray,
+    };
     use arrow_schema::{DataType, Field, Schema};
 
     // Further tests in arrow-integration-testing/tests/ipc_reader.rs
@@ -315,4 +317,61 @@ mod tests {
         let err = decoder.finish().unwrap_err().to_string();
         assert_eq!(err, "Ipc error: Unexpected End of Stream");
     }
+
+    #[test]
+    fn test_read_ree_dict_record_batches_from_buffer() {
+        let schema = Schema::new(vec![Field::new(
+            "test1",
+            DataType::RunEndEncoded(
+                Arc::new(Field::new("run_ends".to_string(), DataType::Int32, false)),
+                Arc::new(Field::new_dict(
+                    "values".to_string(),
+                    DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8)),
+                    true,
+                    0,
+                    false,
+                )),
+            ),
+            true,
+        )]);
+        let batch = RecordBatch::try_new(
+            schema.clone().into(),
+            vec![Arc::new(
+                RunArray::try_new(
+                    &Int32Array::from(vec![1, 2, 3]),
+                    &vec![Some("a"), None, Some("a")]
+                        .into_iter()
+                        .collect::<DictionaryArray<Int32Type>>(),
+                )
+                .expect("Failed to create RunArray"),
+            )],
+        )
+        .expect("Failed to create RecordBatch");
+
+        let mut buffer = vec![];
+        {
+            let mut writer = StreamWriter::try_new_with_options(
+                &mut buffer,
+                &schema,
+                IpcWriteOptions::default().with_preserve_dict_id(false),
+            )
+            .expect("Failed to create StreamWriter");
+            writer.write(&batch).expect("Failed to write RecordBatch");
+            writer.finish().expect("Failed to finish StreamWriter");
+        }
+
+        let mut decoder = StreamDecoder::new();
+        let buf = &mut Buffer::from(buffer.as_slice());
+        while let Some(batch) = decoder
+            .decode(buf)
+            .map_err(|e| {
+                ArrowError::ExternalError(format!("Failed to decode record batch: {}", e).into())
+            })
+            .expect("Failed to decode record batch")
+        {
+            assert_eq!(batch, batch);
+        }
+
+        decoder.finish().expect("Failed to finish decoder");
+    }
 }
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index b09dcdc50..6ef70cdea 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -710,7 +710,9 @@ fn into_zero_offset_run_array<R: RunEndIndexType>(
 }
 
 /// Keeps track of dictionaries that have been written, to avoid emitting the same dictionary
-/// multiple times. Can optionally error if an update to an existing dictionary is attempted, which
+/// multiple times.
+///
+/// Can optionally error if an update to an existing dictionary is attempted, which
 /// isn't allowed in the `FileWriter`.
 pub struct DictionaryTracker {
     written: HashMap<i64, ArrayData>,
diff --git a/arrow-json/Cargo.toml b/arrow-json/Cargo.toml
index dd232f197..517bb03d2 100644
--- a/arrow-json/Cargo.toml
+++ b/arrow-json/Cargo.toml
@@ -45,7 +45,7 @@ num = { version = "0.4", default-features = false, features = ["std"] }
 serde = { version = "1.0", default-features = false }
 serde_json = { version = "1.0", default-features = false, features = ["std"] }
 chrono = { workspace = true }
-lexical-core = { version = "0.8", default-features = false }
+lexical-core = { version = "1.0", default-features = false}
 
 [dev-dependencies]
 tempfile = "3.3"
diff --git a/arrow-schema/src/datatype.rs b/arrow-schema/src/datatype.rs
index 1848c8b3f..b9cfc3d8a 100644
--- a/arrow-schema/src/datatype.rs
+++ b/arrow-schema/src/datatype.rs
@@ -261,9 +261,7 @@ pub enum DataType {
     /// A single LargeBinary array can store up to [`i64::MAX`] bytes
     /// of binary data in total.
     LargeBinary,
-    /// (NOT YET FULLY SUPPORTED) Opaque binary data of variable length.
-    ///
-    /// Note this data type is not yet fully supported. Using it with arrow APIs may result in `panic`s.
+    /// Opaque binary data of variable length.
     ///
     /// Logically the same as [`Self::Binary`], but the internal representation uses a view
     /// struct that contains the string length and either the string's entire data
@@ -280,9 +278,7 @@ pub enum DataType {
     /// A single LargeUtf8 array can store up to [`i64::MAX`] bytes
     /// of string data in total.
     LargeUtf8,
-    /// (NOT YET FULLY SUPPORTED)  A variable-length string in Unicode with UTF-8 encoding
-    ///
-    /// Note this data type is not yet fully supported. Using it with arrow APIs may result in `panic`s.
+    /// A variable-length string in Unicode with UTF-8 encoding
     ///
     /// Logically the same as [`Self::Utf8`], but the internal representation uses a view
     /// struct that contains the string length and either the string's entire data
diff --git a/arrow-schema/src/field.rs b/arrow-schema/src/field.rs
index a84a6ada3..fc4852a3d 100644
--- a/arrow-schema/src/field.rs
+++ b/arrow-schema/src/field.rs
@@ -375,6 +375,7 @@ impl Field {
             | DataType::FixedSizeList(field, _)
             | DataType::Map(field, _) => field.fields(),
             DataType::Dictionary(_, value_field) => Field::_fields(value_field.as_ref()),
+            DataType::RunEndEncoded(_, field) => field.fields(),
             _ => vec![],
         }
     }
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index e07b03d1f..e59ad50dd 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -42,8 +42,9 @@ use arrow_schema::*;
 const FILTER_SLICES_SELECTIVITY_THRESHOLD: f64 = 0.8;
 
 /// An iterator of `(usize, usize)` each representing an interval
-/// `[start, end)` whose slots of a bitmap [Buffer] are true. Each
-/// interval corresponds to a contiguous region of memory to be
+/// `[start, end)` whose slots of a bitmap [Buffer] are true.
+///
+/// Each interval corresponds to a contiguous region of memory to be
 /// "taken" from an array to be filtered.
 ///
 /// ## Notes:
@@ -117,6 +118,7 @@ fn filter_count(filter: &BooleanArray) -> usize {
 pub type Filter<'a> = Box<dyn Fn(&ArrayData) -> ArrayData + 'a>;
 
 /// Returns a prepared function optimized to filter multiple arrays.
+///
 /// Creating this function requires time, but using it is faster than [filter] when the
 /// same filter needs to be applied to multiple arrays (e.g. a multi-column `RecordBatch`).
 /// WARNING: the nulls of `filter` are ignored and the value on its slot is considered.
diff --git a/arrow-string/src/length.rs b/arrow-string/src/length.rs
index 82fb2e0d1..97f876a9f 100644
--- a/arrow-string/src/length.rs
+++ b/arrow-string/src/length.rs
@@ -45,6 +45,7 @@ fn bit_length_impl<P: ArrowPrimitiveType>(
 }
 
 /// Returns an array of Int32/Int64 denoting the length of each value in the array.
+///
 /// For list array, length is the number of elements in each list.
 /// For string array and binary array, length is the number of bytes of each value.
 ///
diff --git a/arrow/examples/README.md b/arrow/examples/README.md
index 7ec3b008b..5c57ec00c 100644
--- a/arrow/examples/README.md
+++ b/arrow/examples/README.md
@@ -25,3 +25,4 @@
 - [`read_csv.rs`](read_csv.rs): Reading CSV files with explicit schema, pretty printing Arrays
 - [`read_csv_infer_schema.rs`](read_csv_infer_schema.rs): Reading CSV files, pretty printing Arrays
 - [`tensor_builder.rs`](tensor_builder.rs): Using tensor builder
+- [`version.rs`](version.rs): Print the arrow version and exit
diff --git a/arrow/examples/version.rs b/arrow/examples/version.rs
new file mode 100644
index 000000000..60dc93343
--- /dev/null
+++ b/arrow/examples/version.rs
@@ -0,0 +1,24 @@
+// 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.
+
+//! Print the arrow version and exit
+
+use arrow::ARROW_VERSION;
+
+fn main() {
+    println!("arrow version: {ARROW_VERSION}");
+}
diff --git a/arrow/src/error.rs b/arrow/src/error.rs
index f7acec0b3..82330f88d 100644
--- a/arrow/src/error.rs
+++ b/arrow/src/error.rs
@@ -19,4 +19,5 @@
 
 pub use arrow_schema::ArrowError;
 
+/// A specialized `Result` type for Arrow operations.
 pub type Result<T> = std::result::Result<T, ArrowError>;
diff --git a/arrow/src/lib.rs b/arrow/src/lib.rs
index 8796caf43..5002e5bf1 100644
--- a/arrow/src/lib.rs
+++ b/arrow/src/lib.rs
@@ -361,15 +361,18 @@
 //! [Apache Parquet]: https://parquet.apache.org/
 //! [DataFusion]: https://github.com/apache/arrow-datafusion
 //! [issue tracker]: https://github.com/apache/arrow-rs/issues
-//!
 
 #![deny(clippy::redundant_clone)]
 #![warn(missing_debug_implementations)]
+#![warn(missing_docs)]
 #![allow(rustdoc::invalid_html_tags)]
 pub use arrow_array::{downcast_dictionary_array, downcast_primitive_array};
 
 pub use arrow_buffer::{alloc, buffer};
 
+/// Arrow crate version
+pub const ARROW_VERSION: &str = env!("CARGO_PKG_VERSION");
+
 pub mod array;
 pub mod compute;
 #[cfg(feature = "csv")]
@@ -387,6 +390,7 @@ pub use arrow_json as json;
 #[cfg(feature = "pyarrow")]
 pub mod pyarrow;
 
+/// Contains the `RecordBatch` type and associated traits
 pub mod record_batch {
     pub use arrow_array::{
         RecordBatch, RecordBatchIterator, RecordBatchOptions, RecordBatchReader, RecordBatchWriter,
diff --git a/arrow/src/pyarrow.rs b/arrow/src/pyarrow.rs
index 336398cbf..b05c967d7 100644
--- a/arrow/src/pyarrow.rs
+++ b/arrow/src/pyarrow.rs
@@ -18,6 +18,7 @@
 //! Pass Arrow objects from and to PyArrow, using Arrow's
 //! [C Data Interface](https://arrow.apache.org/docs/format/CDataInterface.html)
 //! and [pyo3](https://docs.rs/pyo3/latest/pyo3/).
+//!
 //! For underlying implementation, see the [ffi] module.
 //!
 //! One can use these to write Python functions that take and return PyArrow
@@ -76,23 +77,30 @@ use crate::ffi_stream::{ArrowArrayStreamReader, FFI_ArrowArrayStream};
 use crate::record_batch::RecordBatch;
 
 import_exception!(pyarrow, ArrowException);
+/// Represents an exception raised by PyArrow.
 pub type PyArrowException = ArrowException;
 
 fn to_py_err(err: ArrowError) -> PyErr {
     PyArrowException::new_err(err.to_string())
 }
 
+/// Trait for converting Python objects to arrow-rs types.
 pub trait FromPyArrow: Sized {
+    /// Convert a Python object to an arrow-rs type.
+    ///
+    /// Takes a GIL-bound value from Python and returns a result with the arrow-rs type.
     fn from_pyarrow_bound(value: &Bound<PyAny>) -> PyResult<Self>;
 }
 
 /// Create a new PyArrow object from a arrow-rs type.
 pub trait ToPyArrow {
+    /// Convert the implemented type into a Python object without consuming it.
     fn to_pyarrow(&self, py: Python) -> PyResult<PyObject>;
 }
 
 /// Convert an arrow-rs type into a PyArrow object.
 pub trait IntoPyArrow {
+    /// Convert the implemented type into a Python object while consuming it.
     fn into_pyarrow(self, py: Python) -> PyResult<PyObject>;
 }
 
@@ -354,7 +362,7 @@ impl FromPyArrow for RecordBatch {
             validate_pycapsule(array_capsule, "arrow_array")?;
 
             let schema_ptr = unsafe { schema_capsule.reference::<FFI_ArrowSchema>() };
-            let ffi_array = unsafe { FFI_ArrowArray::from_raw(array_capsule.pointer() as _) };
+            let ffi_array = unsafe { FFI_ArrowArray::from_raw(array_capsule.pointer().cast()) };
             let array_data = unsafe { ffi::from_ffi(ffi_array, schema_ptr) }.map_err(to_py_err)?;
             if !matches!(array_data.data_type(), DataType::Struct(_)) {
                 return Err(PyTypeError::new_err(
@@ -472,8 +480,10 @@ impl IntoPyArrow for ArrowArrayStreamReader {
     }
 }
 
-/// A newtype wrapper. When wrapped around a type `T: FromPyArrow`, it
-/// implements `FromPyObject` for the PyArrow objects. When wrapped around a
+/// A newtype wrapper for types implementing [`FromPyArrow`] or [`IntoPyArrow`].
+///
+/// When wrapped around a type `T: FromPyArrow`, it
+/// implements [`FromPyObject`] for the PyArrow objects. When wrapped around a
 /// `T: IntoPyArrow`, it implements `IntoPy<PyObject>` for the wrapped type.
 #[derive(Debug)]
 pub struct PyArrowType<T>(pub T);
diff --git a/arrow/src/tensor.rs b/arrow/src/tensor.rs
index f236e6422..cd135a2f0 100644
--- a/arrow/src/tensor.rs
+++ b/arrow/src/tensor.rs
@@ -80,36 +80,67 @@ pub struct Tensor<'a, T: ArrowPrimitiveType> {
     _marker: PhantomData<T>,
 }
 
+/// [Tensor] of type [BooleanType]
 pub type BooleanTensor<'a> = Tensor<'a, BooleanType>;
+/// [Tensor] of type [Int8Type]
 pub type Date32Tensor<'a> = Tensor<'a, Date32Type>;
+/// [Tensor] of type [Int16Type]
 pub type Date64Tensor<'a> = Tensor<'a, Date64Type>;
+/// [Tensor] of type [Decimal128Type]
 pub type Decimal128Tensor<'a> = Tensor<'a, Decimal128Type>;
+/// [Tensor] of type [Decimal256Type]
 pub type Decimal256Tensor<'a> = Tensor<'a, Decimal256Type>;
+/// [Tensor] of type [DurationMicrosecondType]
 pub type DurationMicrosecondTensor<'a> = Tensor<'a, DurationMicrosecondType>;
+/// [Tensor] of type [DurationMillisecondType]
 pub type DurationMillisecondTensor<'a> = Tensor<'a, DurationMillisecondType>;
+/// [Tensor] of type [DurationNanosecondType]
 pub type DurationNanosecondTensor<'a> = Tensor<'a, DurationNanosecondType>;
+/// [Tensor] of type [DurationSecondType]
 pub type DurationSecondTensor<'a> = Tensor<'a, DurationSecondType>;
+/// [Tensor] of type [Float16Type]
 pub type Float16Tensor<'a> = Tensor<'a, Float16Type>;
+/// [Tensor] of type [Float32Type]
 pub type Float32Tensor<'a> = Tensor<'a, Float32Type>;
+/// [Tensor] of type [Float64Type]
 pub type Float64Tensor<'a> = Tensor<'a, Float64Type>;
+/// [Tensor] of type [Int8Type]
 pub type Int8Tensor<'a> = Tensor<'a, Int8Type>;
+/// [Tensor] of type [Int16Type]
 pub type Int16Tensor<'a> = Tensor<'a, Int16Type>;
+/// [Tensor] of type [Int32Type]
 pub type Int32Tensor<'a> = Tensor<'a, Int32Type>;
+/// [Tensor] of type [Int64Type]
 pub type Int64Tensor<'a> = Tensor<'a, Int64Type>;
+/// [Tensor] of type [IntervalDayTimeType]
 pub type IntervalDayTimeTensor<'a> = Tensor<'a, IntervalDayTimeType>;
+/// [Tensor] of type [IntervalMonthDayNanoType]
 pub type IntervalMonthDayNanoTensor<'a> = Tensor<'a, IntervalMonthDayNanoType>;
+/// [Tensor] of type [IntervalYearMonthType]
 pub type IntervalYearMonthTensor<'a> = Tensor<'a, IntervalYearMonthType>;
+/// [Tensor] of type [Time32MillisecondType]
 pub type Time32MillisecondTensor<'a> = Tensor<'a, Time32MillisecondType>;
+/// [Tensor] of type [Time32SecondType]
 pub type Time32SecondTensor<'a> = Tensor<'a, Time32SecondType>;
+/// [Tensor] of type [Time64MicrosecondType]
 pub type Time64MicrosecondTensor<'a> = Tensor<'a, Time64MicrosecondType>;
+/// [Tensor] of type [Time64NanosecondType]
 pub type Time64NanosecondTensor<'a> = Tensor<'a, Time64NanosecondType>;
+/// [Tensor] of type [TimestampMicrosecondType]
 pub type TimestampMicrosecondTensor<'a> = Tensor<'a, TimestampMicrosecondType>;
+/// [Tensor] of type [TimestampMillisecondType]
 pub type TimestampMillisecondTensor<'a> = Tensor<'a, TimestampMillisecondType>;
+/// [Tensor] of type [TimestampNanosecondType]
 pub type TimestampNanosecondTensor<'a> = Tensor<'a, TimestampNanosecondType>;
+/// [Tensor] of type [TimestampSecondType]
 pub type TimestampSecondTensor<'a> = Tensor<'a, TimestampSecondType>;
+/// [Tensor] of type [UInt8Type]
 pub type UInt8Tensor<'a> = Tensor<'a, UInt8Type>;
+/// [Tensor] of type [UInt16Type]
 pub type UInt16Tensor<'a> = Tensor<'a, UInt16Type>;
+/// [Tensor] of type [UInt32Type]
 pub type UInt32Tensor<'a> = Tensor<'a, UInt32Type>;
+/// [Tensor] of type [UInt64Type]
 pub type UInt64Tensor<'a> = Tensor<'a, UInt64Type>;
 
 impl<'a, T: ArrowPrimitiveType> Tensor<'a, T> {
diff --git a/arrow/src/util/bench_util.rs b/arrow/src/util/bench_util.rs
index 2561c925a..8eaae36db 100644
--- a/arrow/src/util/bench_util.rs
+++ b/arrow/src/util/bench_util.rs
@@ -50,6 +50,8 @@ where
         .collect()
 }
 
+/// Creates a [`PrimitiveArray`] of a given `size` and `null_density`
+/// filling it with random numbers generated using the provided `seed`.
 pub fn create_primitive_array_with_seed<T>(
     size: usize,
     null_density: f32,
@@ -72,6 +74,8 @@ where
         .collect()
 }
 
+/// Creates a [`PrimitiveArray`] of a given `size` and `null_density`
+/// filling it with random [`IntervalMonthDayNano`] generated using the provided `seed`.
 pub fn create_month_day_nano_array_with_seed(
     size: usize,
     null_density: f32,
@@ -108,7 +112,9 @@ where
         .collect()
 }
 
-/// Creates a random (but fixed-seeded) string array of a given size and null density, strings have a random length
+/// Creates a random (but fixed-seeded) string array of a given size and null density.
+///
+/// Strings have a random length
 /// between 0 and 400 alphanumeric characters. `0..400` is chosen to cover a wide range of common string lengths,
 /// which have a dramatic impact on performance of some queries, e.g. LIKE/ILIKE/regex.
 pub fn create_string_array<Offset: OffsetSizeTrait>(
diff --git a/arrow/src/util/data_gen.rs b/arrow/src/util/data_gen.rs
index edb406c0f..55cab368a 100644
--- a/arrow/src/util/data_gen.rs
+++ b/arrow/src/util/data_gen.rs
@@ -394,8 +394,10 @@ fn create_random_null_buffer(size: usize, null_density: f32) -> Buffer {
 /// Useful for testing. The range of values are not likely to be representative of the
 /// actual bounds.
 pub trait RandomTemporalValue: ArrowTemporalType {
+    /// Returns the range of values for `impl`'d type
     fn value_range() -> impl SampleRange<Self::Native>;
 
+    /// Generate a random value within the range of the type
     fn gen_range<R: Rng>(rng: &mut R) -> Self::Native
     where
         Self::Native: SampleUniform,
@@ -403,6 +405,7 @@ pub trait RandomTemporalValue: ArrowTemporalType {
         rng.gen_range(Self::value_range())
     }
 
+    /// Generate a random value of the type
     fn random<R: Rng>(rng: &mut R) -> Self::Native
     where
         Self::Native: SampleUniform,
diff --git a/arrow/src/util/mod.rs b/arrow/src/util/mod.rs
index 7f7257100..2c131669b 100644
--- a/arrow/src/util/mod.rs
+++ b/arrow/src/util/mod.rs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+//! Utility functions for working with Arrow data
 pub use arrow_buffer::{bit_chunk_iterator, bit_util};
 
 pub use arrow_data::bit_iterator;
diff --git a/arrow/src/util/string_writer.rs b/arrow/src/util/string_writer.rs
index 57a98d6f5..5d8725d35 100644
--- a/arrow/src/util/string_writer.rs
+++ b/arrow/src/util/string_writer.rs
@@ -63,27 +63,24 @@
 //! }
 //! ```
 
+use core::str;
 use std::fmt::Formatter;
 use std::io::{Error, ErrorKind, Result, Write};
 
-#[derive(Debug)]
+/// A writer that allows writing to a `String`
+/// like an `std::io::Write` object.
+#[derive(Debug, Default)]
 pub struct StringWriter {
     data: String,
 }
 
 impl StringWriter {
+    /// Create a new `StringWriter`
     pub fn new() -> Self {
-        StringWriter {
-            data: String::new(),
-        }
+        Self::default()
     }
 }
 
-impl Default for StringWriter {
-    fn default() -> Self {
-        Self::new()
-    }
-}
 impl std::fmt::Display for StringWriter {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.data)
@@ -92,13 +89,13 @@ impl std::fmt::Display for StringWriter {
 
 impl Write for StringWriter {
     fn write(&mut self, buf: &[u8]) -> Result<usize> {
-        let string = match String::from_utf8(buf.to_vec()) {
+        let string = match str::from_utf8(buf) {
             Ok(x) => x,
             Err(e) => {
                 return Err(Error::new(ErrorKind::InvalidData, e));
             }
         };
-        self.data.push_str(&string);
+        self.data.push_str(string);
         Ok(string.len())
     }
 
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 4a773e7a1..a27ed0533 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -77,7 +77,7 @@ use crate::client::parts::Parts;
 pub use credential::{AwsAuthorizer, AwsCredential};
 
 /// Interface for [Amazon S3](https://aws.amazon.com/s3/).
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct AmazonS3 {
     client: Arc<S3Client>,
 }
diff --git a/object_store/src/azure/builder.rs b/object_store/src/azure/builder.rs
index 0208073e8..35cedeafc 100644
--- a/object_store/src/azure/builder.rs
+++ b/object_store/src/azure/builder.rs
@@ -17,8 +17,8 @@
 
 use crate::azure::client::{AzureClient, AzureConfig};
 use crate::azure::credential::{
-    AzureAccessKey, AzureCliCredential, ClientSecretOAuthProvider, ImdsManagedIdentityProvider,
-    WorkloadIdentityOAuthProvider,
+    AzureAccessKey, AzureCliCredential, ClientSecretOAuthProvider, FabricTokenOAuthProvider,
+    ImdsManagedIdentityProvider, WorkloadIdentityOAuthProvider,
 };
 use crate::azure::{AzureCredential, AzureCredentialProvider, MicrosoftAzure, STORE};
 use crate::client::TokenCredentialProvider;
@@ -172,6 +172,14 @@ pub struct MicrosoftAzureBuilder {
     use_fabric_endpoint: ConfigValue<bool>,
     /// When set to true, skips tagging objects
     disable_tagging: ConfigValue<bool>,
+    /// Fabric token service url
+    fabric_token_service_url: Option<String>,
+    /// Fabric workload host
+    fabric_workload_host: Option<String>,
+    /// Fabric session token
+    fabric_session_token: Option<String>,
+    /// Fabric cluster identifier
+    fabric_cluster_identifier: Option<String>,
 }
 
 /// Configuration keys for [`MicrosoftAzureBuilder`]
@@ -336,6 +344,34 @@ pub enum AzureConfigKey {
     /// - `disable_tagging`
     DisableTagging,
 
+    /// Fabric token service url
+    ///
+    /// Supported keys:
+    /// - `azure_fabric_token_service_url`
+    /// - `fabric_token_service_url`
+    FabricTokenServiceUrl,
+
+    /// Fabric workload host
+    ///
+    /// Supported keys:
+    /// - `azure_fabric_workload_host`
+    /// - `fabric_workload_host`
+    FabricWorkloadHost,
+
+    /// Fabric session token
+    ///
+    /// Supported keys:
+    /// - `azure_fabric_session_token`
+    /// - `fabric_session_token`
+    FabricSessionToken,
+
+    /// Fabric cluster identifier
+    ///
+    /// Supported keys:
+    /// - `azure_fabric_cluster_identifier`
+    /// - `fabric_cluster_identifier`
+    FabricClusterIdentifier,
+
     /// Client options
     Client(ClientConfigKey),
 }
@@ -361,6 +397,10 @@ impl AsRef<str> for AzureConfigKey {
             Self::SkipSignature => "azure_skip_signature",
             Self::ContainerName => "azure_container_name",
             Self::DisableTagging => "azure_disable_tagging",
+            Self::FabricTokenServiceUrl => "azure_fabric_token_service_url",
+            Self::FabricWorkloadHost => "azure_fabric_workload_host",
+            Self::FabricSessionToken => "azure_fabric_session_token",
+            Self::FabricClusterIdentifier => "azure_fabric_cluster_identifier",
             Self::Client(key) => key.as_ref(),
         }
     }
@@ -406,6 +446,14 @@ impl FromStr for AzureConfigKey {
             "azure_skip_signature" | "skip_signature" => Ok(Self::SkipSignature),
             "azure_container_name" | "container_name" => Ok(Self::ContainerName),
             "azure_disable_tagging" | "disable_tagging" => Ok(Self::DisableTagging),
+            "azure_fabric_token_service_url" | "fabric_token_service_url" => {
+                Ok(Self::FabricTokenServiceUrl)
+            }
+            "azure_fabric_workload_host" | "fabric_workload_host" => Ok(Self::FabricWorkloadHost),
+            "azure_fabric_session_token" | "fabric_session_token" => Ok(Self::FabricSessionToken),
+            "azure_fabric_cluster_identifier" | "fabric_cluster_identifier" => {
+                Ok(Self::FabricClusterIdentifier)
+            }
             // Backwards compatibility
             "azure_allow_http" => Ok(Self::Client(ClientConfigKey::AllowHttp)),
             _ => match s.strip_prefix("azure_").unwrap_or(s).parse() {
@@ -525,6 +573,14 @@ impl MicrosoftAzureBuilder {
             }
             AzureConfigKey::ContainerName => self.container_name = Some(value.into()),
             AzureConfigKey::DisableTagging => self.disable_tagging.parse(value),
+            AzureConfigKey::FabricTokenServiceUrl => {
+                self.fabric_token_service_url = Some(value.into())
+            }
+            AzureConfigKey::FabricWorkloadHost => self.fabric_workload_host = Some(value.into()),
+            AzureConfigKey::FabricSessionToken => self.fabric_session_token = Some(value.into()),
+            AzureConfigKey::FabricClusterIdentifier => {
+                self.fabric_cluster_identifier = Some(value.into())
+            }
         };
         self
     }
@@ -561,6 +617,10 @@ impl MicrosoftAzureBuilder {
             AzureConfigKey::Client(key) => self.client_options.get_config_value(key),
             AzureConfigKey::ContainerName => self.container_name.clone(),
             AzureConfigKey::DisableTagging => Some(self.disable_tagging.to_string()),
+            AzureConfigKey::FabricTokenServiceUrl => self.fabric_token_service_url.clone(),
+            AzureConfigKey::FabricWorkloadHost => self.fabric_workload_host.clone(),
+            AzureConfigKey::FabricSessionToken => self.fabric_session_token.clone(),
+            AzureConfigKey::FabricClusterIdentifier => self.fabric_cluster_identifier.clone(),
         }
     }
 
@@ -856,6 +916,30 @@ impl MicrosoftAzureBuilder {
 
             let credential = if let Some(credential) = self.credentials {
                 credential
+            } else if let (
+                Some(fabric_token_service_url),
+                Some(fabric_workload_host),
+                Some(fabric_session_token),
+                Some(fabric_cluster_identifier),
+            ) = (
+                &self.fabric_token_service_url,
+                &self.fabric_workload_host,
+                &self.fabric_session_token,
+                &self.fabric_cluster_identifier,
+            ) {
+                // This case should precede the bearer token case because it is more specific and will utilize the bearer token.
+                let fabric_credential = FabricTokenOAuthProvider::new(
+                    fabric_token_service_url,
+                    fabric_workload_host,
+                    fabric_session_token,
+                    fabric_cluster_identifier,
+                    self.bearer_token.clone(),
+                );
+                Arc::new(TokenCredentialProvider::new(
+                    fabric_credential,
+                    self.client_options.client()?,
+                    self.retry_config.clone(),
+                )) as _
             } else if let Some(bearer_token) = self.bearer_token {
                 static_creds(AzureCredential::BearerToken(bearer_token))
             } else if let Some(access_key) = self.access_key {
diff --git a/object_store/src/azure/credential.rs b/object_store/src/azure/credential.rs
index 7808c7c4a..6b5fa19d1 100644
--- a/object_store/src/azure/credential.rs
+++ b/object_store/src/azure/credential.rs
@@ -22,7 +22,7 @@ use crate::client::{CredentialProvider, TokenProvider};
 use crate::util::hmac_sha256;
 use crate::RetryConfig;
 use async_trait::async_trait;
-use base64::prelude::BASE64_STANDARD;
+use base64::prelude::{BASE64_STANDARD, BASE64_URL_SAFE_NO_PAD};
 use base64::Engine;
 use chrono::{DateTime, SecondsFormat, Utc};
 use reqwest::header::{
@@ -51,10 +51,15 @@ pub(crate) static BLOB_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-typ
 pub(crate) static DELETE_SNAPSHOTS: HeaderName = HeaderName::from_static("x-ms-delete-snapshots");
 pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
 static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+static PARTNER_TOKEN: HeaderName = HeaderName::from_static("x-ms-partner-token");
+static CLUSTER_IDENTIFIER: HeaderName = HeaderName::from_static("x-ms-cluster-identifier");
+static WORKLOAD_RESOURCE: HeaderName = HeaderName::from_static("x-ms-workload-resource-moniker");
+static PROXY_HOST: HeaderName = HeaderName::from_static("x-ms-proxy-host");
 pub(crate) const RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
 const CONTENT_TYPE_JSON: &str = "application/json";
 const MSI_SECRET_ENV_KEY: &str = "IDENTITY_HEADER";
 const MSI_API_VERSION: &str = "2019-08-01";
+const TOKEN_MIN_TTL: u64 = 300;
 
 /// OIDC scope used when interacting with OAuth2 APIs
 ///
@@ -934,6 +939,113 @@ impl AzureCliCredential {
     }
 }
 
+/// Encapsulates the logic to perform an OAuth token challenge for Fabric
+#[derive(Debug)]
+pub struct FabricTokenOAuthProvider {
+    fabric_token_service_url: String,
+    fabric_workload_host: String,
+    fabric_session_token: String,
+    fabric_cluster_identifier: String,
+    storage_access_token: Option<String>,
+    token_expiry: Option<u64>,
+}
+
+#[derive(Debug, Deserialize)]
+struct Claims {
+    exp: u64,
+}
+
+impl FabricTokenOAuthProvider {
+    /// Create a new [`FabricTokenOAuthProvider`] for an azure backed store
+    pub fn new(
+        fabric_token_service_url: impl Into<String>,
+        fabric_workload_host: impl Into<String>,
+        fabric_session_token: impl Into<String>,
+        fabric_cluster_identifier: impl Into<String>,
+        storage_access_token: Option<String>,
+    ) -> Self {
+        let (storage_access_token, token_expiry) = match storage_access_token {
+            Some(token) => match Self::validate_and_get_expiry(&token) {
+                Some(expiry) if expiry > Self::get_current_timestamp() + TOKEN_MIN_TTL => {
+                    (Some(token), Some(expiry))
+                }
+                _ => (None, None),
+            },
+            None => (None, None),
+        };
+
+        Self {
+            fabric_token_service_url: fabric_token_service_url.into(),
+            fabric_workload_host: fabric_workload_host.into(),
+            fabric_session_token: fabric_session_token.into(),
+            fabric_cluster_identifier: fabric_cluster_identifier.into(),
+            storage_access_token,
+            token_expiry,
+        }
+    }
+
+    fn validate_and_get_expiry(token: &str) -> Option<u64> {
+        let payload = token.split('.').nth(1)?;
+        let decoded_bytes = BASE64_URL_SAFE_NO_PAD.decode(payload).ok()?;
+        let decoded_str = str::from_utf8(&decoded_bytes).ok()?;
+        let claims: Claims = serde_json::from_str(decoded_str).ok()?;
+        Some(claims.exp)
+    }
+
+    fn get_current_timestamp() -> u64 {
+        SystemTime::now()
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .map_or(0, |d| d.as_secs())
+    }
+}
+
+#[async_trait::async_trait]
+impl TokenProvider for FabricTokenOAuthProvider {
+    type Credential = AzureCredential;
+
+    /// Fetch a token
+    async fn fetch_token(
+        &self,
+        client: &Client,
+        retry: &RetryConfig,
+    ) -> crate::Result<TemporaryToken<Arc<AzureCredential>>> {
+        if let Some(storage_access_token) = &self.storage_access_token {
+            if let Some(expiry) = self.token_expiry {
+                let exp_in = expiry - Self::get_current_timestamp();
+                if exp_in > TOKEN_MIN_TTL {
+                    return Ok(TemporaryToken {
+                        token: Arc::new(AzureCredential::BearerToken(storage_access_token.clone())),
+                        expiry: Some(Instant::now() + Duration::from_secs(exp_in)),
+                    });
+                }
+            }
+        }
+
+        let query_items = vec![("resource", AZURE_STORAGE_RESOURCE)];
+        let access_token: String = client
+            .request(Method::GET, &self.fabric_token_service_url)
+            .header(&PARTNER_TOKEN, self.fabric_session_token.as_str())
+            .header(&CLUSTER_IDENTIFIER, self.fabric_cluster_identifier.as_str())
+            .header(&WORKLOAD_RESOURCE, self.fabric_cluster_identifier.as_str())
+            .header(&PROXY_HOST, self.fabric_workload_host.as_str())
+            .query(&query_items)
+            .retryable(retry)
+            .idempotent(true)
+            .send()
+            .await
+            .context(TokenRequestSnafu)?
+            .text()
+            .await
+            .context(TokenResponseBodySnafu)?;
+        let exp_in = Self::validate_and_get_expiry(&access_token)
+            .map_or(3600, |expiry| expiry - Self::get_current_timestamp());
+        Ok(TemporaryToken {
+            token: Arc::new(AzureCredential::BearerToken(access_token)),
+            expiry: Some(Instant::now() + Duration::from_secs(exp_in)),
+        })
+    }
+}
+
 #[async_trait]
 impl CredentialProvider for AzureCliCredential {
     type Credential = AzureCredential;
diff --git a/parquet/src/arrow/array_reader/mod.rs b/parquet/src/arrow/array_reader/mod.rs
index fef1e8722..a5ea426e9 100644
--- a/parquet/src/arrow/array_reader/mod.rs
+++ b/parquet/src/arrow/array_reader/mod.rs
@@ -61,12 +61,16 @@ pub use struct_array::StructArrayReader;
 
 /// Array reader reads parquet data into arrow array.
 pub trait ArrayReader: Send {
+    // TODO: this function is never used, and the trait is not public. Perhaps this should be
+    // removed.
+    #[allow(dead_code)]
     fn as_any(&self) -> &dyn Any;
 
     /// Returns the arrow type of this array reader.
     fn get_data_type(&self) -> &ArrowType;
 
     /// Reads at most `batch_size` records into an arrow array and return it.
+    #[cfg(any(feature = "experimental", test))]
     fn next_batch(&mut self, batch_size: usize) -> Result<ArrayRef> {
         self.read_records(batch_size)?;
         self.consume_batch()
diff --git a/parquet/src/arrow/arrow_reader/selection.rs b/parquet/src/arrow/arrow_reader/selection.rs
index ce3fbbf4f..378d2253f 100644
--- a/parquet/src/arrow/arrow_reader/selection.rs
+++ b/parquet/src/arrow/arrow_reader/selection.rs
@@ -436,6 +436,11 @@ impl RowSelection {
     pub fn row_count(&self) -> usize {
         self.iter().filter(|s| !s.skip).map(|s| s.row_count).sum()
     }
+
+    /// Returns the number of de-selected rows
+    pub fn skipped_row_count(&self) -> usize {
+        self.iter().filter(|s| s.skip).map(|s| s.row_count).sum()
+    }
 }
 
 impl From<Vec<RowSelector>> for RowSelection {
@@ -1345,4 +1350,32 @@ mod tests {
             ]
         );
     }
+
+    #[test]
+    fn test_row_count() {
+        let selection = RowSelection::from(vec![
+            RowSelector::skip(34),
+            RowSelector::select(12),
+            RowSelector::skip(3),
+            RowSelector::select(35),
+        ]);
+
+        assert_eq!(selection.row_count(), 12 + 35);
+        assert_eq!(selection.skipped_row_count(), 34 + 3);
+
+        let selection = RowSelection::from(vec![RowSelector::select(12), RowSelector::select(35)]);
+
+        assert_eq!(selection.row_count(), 12 + 35);
+        assert_eq!(selection.skipped_row_count(), 0);
+
+        let selection = RowSelection::from(vec![RowSelector::skip(34), RowSelector::skip(3)]);
+
+        assert_eq!(selection.row_count(), 0);
+        assert_eq!(selection.skipped_row_count(), 34 + 3);
+
+        let selection = RowSelection::from(vec![]);
+
+        assert_eq!(selection.row_count(), 0);
+        assert_eq!(selection.skipped_row_count(), 0);
+    }
 }
diff --git a/parquet/src/arrow/async_reader/metadata.rs b/parquet/src/arrow/async_reader/metadata.rs
index 9224ea3f6..b4bf77f26 100644
--- a/parquet/src/arrow/async_reader/metadata.rs
+++ b/parquet/src/arrow/async_reader/metadata.rs
@@ -21,6 +21,7 @@ use crate::file::footer::{decode_footer, decode_metadata};
 use crate::file::metadata::ParquetMetaData;
 use crate::file::page_index::index::Index;
 use crate::file::page_index::index_reader::{acc_range, decode_column_index, decode_offset_index};
+use crate::file::FOOTER_SIZE;
 use bytes::Bytes;
 use futures::future::BoxFuture;
 use futures::FutureExt;
@@ -53,7 +54,7 @@ impl<F: MetadataFetch> MetadataLoader<F> {
     ///
     /// See [`fetch_parquet_metadata`] for the meaning of the individual parameters
     pub async fn load(mut fetch: F, file_size: usize, prefetch: Option<usize>) -> Result<Self> {
-        if file_size < 8 {
+        if file_size < FOOTER_SIZE {
             return Err(ParquetError::EOF(format!(
                 "file size of {file_size} is less than footer"
             )));
@@ -62,20 +63,22 @@ impl<F: MetadataFetch> MetadataLoader<F> {
         // If a size hint is provided, read more than the minimum size
         // to try and avoid a second fetch.
         let footer_start = if let Some(size_hint) = prefetch {
+            // check for hint smaller than footer
+            let size_hint = std::cmp::max(size_hint, FOOTER_SIZE);
             file_size.saturating_sub(size_hint)
         } else {
-            file_size - 8
+            file_size - FOOTER_SIZE
         };
 
         let suffix = fetch.fetch(footer_start..file_size).await?;
         let suffix_len = suffix.len();
 
-        let mut footer = [0; 8];
-        footer.copy_from_slice(&suffix[suffix_len - 8..suffix_len]);
+        let mut footer = [0; FOOTER_SIZE];
+        footer.copy_from_slice(&suffix[suffix_len - FOOTER_SIZE..suffix_len]);
 
         let length = decode_footer(&footer)?;
 
-        if file_size < length + 8 {
+        if file_size < length + FOOTER_SIZE {
             return Err(ParquetError::EOF(format!(
                 "file size of {} is less than footer + metadata {}",
                 file_size,
@@ -84,14 +87,14 @@ impl<F: MetadataFetch> MetadataLoader<F> {
         }
 
         // Did not fetch the entire file metadata in the initial read, need to make a second request
-        let (metadata, remainder) = if length > suffix_len - 8 {
-            let metadata_start = file_size - length - 8;
-            let meta = fetch.fetch(metadata_start..file_size - 8).await?;
+        let (metadata, remainder) = if length > suffix_len - FOOTER_SIZE {
+            let metadata_start = file_size - length - FOOTER_SIZE;
+            let meta = fetch.fetch(metadata_start..file_size - FOOTER_SIZE).await?;
             (decode_metadata(&meta)?, None)
         } else {
-            let metadata_start = file_size - length - 8 - footer_start;
+            let metadata_start = file_size - length - FOOTER_SIZE - footer_start;
 
-            let slice = &suffix[metadata_start..suffix_len - 8];
+            let slice = &suffix[metadata_start..suffix_len - FOOTER_SIZE];
             (
                 decode_metadata(slice)?,
                 Some((footer_start, suffix.slice(..metadata_start))),
@@ -273,6 +276,14 @@ mod tests {
         assert_eq!(actual.file_metadata().schema(), expected);
         assert_eq!(fetch_count.load(Ordering::SeqCst), 2);
 
+        // Metadata hint too small - below footer size
+        fetch_count.store(0, Ordering::SeqCst);
+        let actual = fetch_parquet_metadata(&mut fetch, len, Some(7))
+            .await
+            .unwrap();
+        assert_eq!(actual.file_metadata().schema(), expected);
+        assert_eq!(fetch_count.load(Ordering::SeqCst), 2);
+
         // Metadata hint too small
         fetch_count.store(0, Ordering::SeqCst);
         let actual = fetch_parquet_metadata(&mut fetch, len, Some(10))
diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs
index 02c2f44f6..8fde542f5 100644
--- a/parquet/src/basic.rs
+++ b/parquet/src/basic.rs
@@ -39,6 +39,7 @@ pub use crate::format::{
 // Mirrors `parquet::Type`
 
 /// Types supported by Parquet.
+///
 /// These physical types are intended to be used in combination with the encodings to
 /// control the on disk storage format.
 /// For example INT16 is not included as a type since a good encoding of INT32
@@ -60,6 +61,7 @@ pub enum Type {
 // Mirrors `parquet::ConvertedType`
 
 /// Common types (converted types) used by frameworks when using Parquet.
+///
 /// This helps map between types in those frameworks to the base types in Parquet.
 /// This is only metadata and not needed to read or write the data.
 ///
diff --git a/parquet/src/bloom_filter/mod.rs b/parquet/src/bloom_filter/mod.rs
index a8d68d4b6..f98111416 100644
--- a/parquet/src/bloom_filter/mod.rs
+++ b/parquet/src/bloom_filter/mod.rs
@@ -181,9 +181,10 @@ impl std::ops::IndexMut<usize> for Block {
     }
 }
 
-/// A split block Bloom filter. The creation of this structure is based on the
-/// [`crate::file::properties::BloomFilterProperties`] struct set via [`crate::file::properties::WriterProperties`] and
-/// is thus hidden by default.
+/// A split block Bloom filter.
+///
+/// The creation of this structure is based on the [`crate::file::properties::BloomFilterProperties`]
+/// struct set via [`crate::file::properties::WriterProperties`] and is thus hidden by default.
 #[derive(Debug, Clone)]
 pub struct Sbbf(Vec<Block>);
 
diff --git a/parquet/src/encodings/decoding.rs b/parquet/src/encodings/decoding.rs
index b5217d02f..e7f437304 100644
--- a/parquet/src/encodings/decoding.rs
+++ b/parquet/src/encodings/decoding.rs
@@ -273,6 +273,7 @@ pub struct PlainDecoderDetails {
 }
 
 /// Plain decoding that supports all types.
+///
 /// Values are encoded back to back. For native types, data is encoded as little endian.
 /// Floating point types are encoded in IEEE.
 /// See [`PlainEncoder`](crate::encoding::PlainEncoder) for more information.
@@ -333,6 +334,7 @@ impl<T: DataType> Decoder<T> for PlainDecoder<T> {
 // RLE_DICTIONARY/PLAIN_DICTIONARY Decoding
 
 /// Dictionary decoder.
+///
 /// The dictionary encoding builds a dictionary of values encountered in a given column.
 /// The dictionary is be stored in a dictionary page per column chunk.
 /// See [`DictEncoder`](crate::encoding::DictEncoder) for more information.
@@ -824,6 +826,7 @@ where
 // DELTA_LENGTH_BYTE_ARRAY Decoding
 
 /// Delta length byte array decoder.
+///
 /// Only applied to byte arrays to separate the length values and the data, the lengths
 /// are encoded using DELTA_BINARY_PACKED encoding.
 /// See [`DeltaLengthByteArrayEncoder`](crate::encoding::DeltaLengthByteArrayEncoder)
@@ -952,6 +955,7 @@ impl<T: DataType> Decoder<T> for DeltaLengthByteArrayDecoder<T> {
 // DELTA_BYTE_ARRAY Decoding
 
 /// Delta byte array decoder.
+///
 /// Prefix lengths are encoded using `DELTA_BINARY_PACKED` encoding, Suffixes are stored
 /// using `DELTA_LENGTH_BYTE_ARRAY` encoding.
 /// See [`DeltaByteArrayEncoder`](crate::encoding::DeltaByteArrayEncoder) for more
diff --git a/parquet/src/file/writer.rs b/parquet/src/file/writer.rs
index a23a2a332..7b7bfa19c 100644
--- a/parquet/src/file/writer.rs
+++ b/parquet/src/file/writer.rs
@@ -190,14 +190,25 @@ impl<W: Write + Send> SerializedFileWriter<W> {
     /// Creates new row group from this file writer.
     /// In case of IO error or Thrift error, returns `Err`.
     ///
-    /// There is no limit on a number of row groups in a file; however, row groups have
+    /// There can be at most 2^15 row groups in a file; and row groups have
     /// to be written sequentially. Every time the next row group is requested, the
     /// previous row group must be finalised and closed using `RowGroupWriter::close` method.
     pub fn next_row_group(&mut self) -> Result<SerializedRowGroupWriter<'_, W>> {
         self.assert_previous_writer_closed()?;
         let ordinal = self.row_group_index;
 
-        self.row_group_index += 1;
+        let ordinal: i16 = ordinal.try_into().map_err(|_| {
+            ParquetError::General(format!(
+                "Parquet does not support more than {} row groups per file (currently: {})",
+                i16::MAX,
+                ordinal
+            ))
+        })?;
+
+        self.row_group_index = self
+            .row_group_index
+            .checked_add(1)
+            .expect("SerializedFileWriter::row_group_index overflowed");
 
         let bloom_filter_position = self.properties().bloom_filter_position();
         let row_groups = &mut self.row_groups;
@@ -227,7 +238,7 @@ impl<W: Write + Send> SerializedFileWriter<W> {
             self.descr.clone(),
             self.props.clone(),
             &mut self.buf,
-            ordinal as i16,
+            ordinal,
             Some(Box::new(on_close)),
         );
         Ok(row_group_writer)
diff --git a/parquet/src/lib.rs b/parquet/src/lib.rs
index 543c629d3..a54d4a427 100644
--- a/parquet/src/lib.rs
+++ b/parquet/src/lib.rs
@@ -116,8 +116,8 @@ pub mod basic;
 ///
 /// [parquet.thrift]: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift
 // see parquet/CONTRIBUTING.md for instructions on regenerating
-#[allow(clippy::derivable_impls, clippy::match_single_binding)]
-// Don't try and format auto generated code
+// Don't try clippy and format auto generated code
+#[allow(clippy::all)]
 #[rustfmt::skip]
 pub mod format;
 
diff --git a/parquet/src/record/record_reader.rs b/parquet/src/record/record_reader.rs
index bcfeb95dc..cfaf14a3d 100644
--- a/parquet/src/record/record_reader.rs
+++ b/parquet/src/record/record_reader.rs
@@ -18,7 +18,8 @@
 use super::super::errors::ParquetError;
 use super::super::file::reader::RowGroupReader;
 
-/// read up to `max_records` records from `row_group_reader` into `self`
+/// Read up to `max_records` records from `row_group_reader` into `self`.
+///
 /// The type parameter `T` is used to work around the rust orphan rule
 /// when implementing on types such as `Vec<T>`.
 pub trait RecordReader<T> {
diff --git a/parquet/src/record/record_writer.rs b/parquet/src/record/record_writer.rs
index 0b2b95ef7..56e0aa490 100644
--- a/parquet/src/record/record_writer.rs
+++ b/parquet/src/record/record_writer.rs
@@ -20,16 +20,23 @@ use crate::schema::types::TypePtr;
 use super::super::errors::ParquetError;
 use super::super::file::writer::SerializedRowGroupWriter;
 
-/// `write_to_row_group` writes from `self` into `row_group_writer`
-/// `schema` builds the schema used by `row_group_writer`
+/// Trait describing how to write a record (the implementator) to a row group writer.
+///
+/// [`parquet_derive`] crate provides a derive macro [`ParquetRecordWriter`] for this trait
+/// for unnested structs.
+///
 /// The type parameter `T` is used to work around the rust orphan rule
 /// when implementing on types such as `&[T]`.
+///
+/// [`parquet_derive`]: https://crates.io/crates/parquet_derive
+/// [`ParquetRecordWriter`]: https://docs.rs/parquet_derive/53.0.0/parquet_derive/derive.ParquetRecordWriter.html
 pub trait RecordWriter<T> {
+    /// Writes from `self` into `row_group_writer`.
     fn write_to_row_group<W: std::io::Write + Send>(
         &self,
         row_group_writer: &mut SerializedRowGroupWriter<W>,
     ) -> Result<(), ParquetError>;
 
-    /// Generated schema
+    /// Generated schema used by `row_group_writer`
     fn schema(&self) -> Result<TypePtr, ParquetError>;
 }
diff --git a/parquet_derive/src/lib.rs b/parquet_derive/src/lib.rs
index 9c93e2cca..038d8fa44 100644
--- a/parquet_derive/src/lib.rs
+++ b/parquet_derive/src/lib.rs
@@ -29,8 +29,9 @@ use ::syn::{parse_macro_input, Data, DataStruct, DeriveInput};
 
 mod parquet_field;
 
-/// Derive flat, simple RecordWriter implementations. Works by parsing
-/// a struct tagged with `#[derive(ParquetRecordWriter)]` and emitting
+/// Derive flat, simple RecordWriter implementations.
+///
+/// Works by parsing a struct tagged with `#[derive(ParquetRecordWriter)]` and emitting
 /// the correct writing code for each field of the struct. Column writers
 /// are generated in the order they are defined.
 ///
@@ -143,8 +144,9 @@ pub fn parquet_record_writer(input: proc_macro::TokenStream) -> proc_macro::Toke
   }).into()
 }
 
-/// Derive flat, simple RecordReader implementations. Works by parsing
-/// a struct tagged with `#[derive(ParquetRecordReader)]` and emitting
+/// Derive flat, simple RecordReader implementations.
+///
+/// Works by parsing a struct tagged with `#[derive(ParquetRecordReader)]` and emitting
 /// the correct writing code for each field of the struct. Column readers
 /// are generated by matching names in the schema to the names in the struct.
 ///


Test patch:
diff --git a/arrow/tests/pyarrow.rs b/arrow/tests/pyarrow.rs
index a1c365c31..d9ebd0daa 100644
--- a/arrow/tests/pyarrow.rs
+++ b/arrow/tests/pyarrow.rs
@@ -18,6 +18,8 @@
 use arrow::array::{ArrayRef, Int32Array, StringArray};
 use arrow::pyarrow::{FromPyArrow, ToPyArrow};
 use arrow::record_batch::RecordBatch;
+use arrow_array::builder::{BinaryViewBuilder, StringViewBuilder};
+use arrow_array::{Array, BinaryViewArray, StringViewArray};
 use pyo3::Python;
 use std::sync::Arc;
 
@@ -27,7 +29,9 @@ fn test_to_pyarrow() {
 
     let a: ArrayRef = Arc::new(Int32Array::from(vec![1, 2]));
     let b: ArrayRef = Arc::new(StringArray::from(vec!["a", "b"]));
-    let input = RecordBatch::try_from_iter(vec![("a", a), ("b", b)]).unwrap();
+    // The "very long string" will not be inlined, and force the creation of a data buffer.
+    let c: ArrayRef = Arc::new(StringViewArray::from(vec!["short", "a very long string"]));
+    let input = RecordBatch::try_from_iter(vec![("a", a), ("b", b), ("c", c)]).unwrap();
     println!("input: {:?}", input);
 
     let res = Python::with_gil(|py| {
@@ -40,3 +44,66 @@ fn test_to_pyarrow() {
 
     assert_eq!(input, res);
 }
+
+#[test]
+fn test_to_pyarrow_byte_view() {
+    pyo3::prepare_freethreaded_python();
+
+    for num_variadic_buffers in 0..=2 {
+        let string_view: ArrayRef = Arc::new(string_view_column(num_variadic_buffers));
+        let binary_view: ArrayRef = Arc::new(binary_view_column(num_variadic_buffers));
+
+        let input = RecordBatch::try_from_iter(vec![
+            ("string_view", string_view),
+            ("binary_view", binary_view),
+        ])
+        .unwrap();
+
+        println!("input: {:?}", input);
+        let res = Python::with_gil(|py| {
+            let py_input = input.to_pyarrow(py)?;
+            let records = RecordBatch::from_pyarrow_bound(py_input.bind(py))?;
+            let py_records = records.to_pyarrow(py)?;
+            RecordBatch::from_pyarrow_bound(py_records.bind(py))
+        })
+        .unwrap();
+
+        assert_eq!(input, res);
+    }
+}
+
+fn binary_view_column(num_variadic_buffers: usize) -> BinaryViewArray {
+    let long_scalar = b"but soft what light through yonder window breaks".as_slice();
+    let mut builder = BinaryViewBuilder::new().with_fixed_block_size(long_scalar.len() as u32);
+    // Make sure there is at least one non-inlined value.
+    builder.append_value("inlined".as_bytes());
+
+    for _ in 0..num_variadic_buffers {
+        builder.append_value(long_scalar);
+    }
+
+    let result = builder.finish();
+
+    assert_eq!(result.data_buffers().len(), num_variadic_buffers);
+    assert_eq!(result.len(), num_variadic_buffers + 1);
+
+    result
+}
+
+fn string_view_column(num_variadic_buffers: usize) -> StringViewArray {
+    let long_scalar = "but soft what light through yonder window breaks";
+    let mut builder = StringViewBuilder::new().with_fixed_block_size(long_scalar.len() as u32);
+    // Make sure there is at least one non-inlined value.
+    builder.append_value("inlined");
+
+    for _ in 0..num_variadic_buffers {
+        builder.append_value(long_scalar);
+    }
+
+    let result = builder.finish();
+
+    assert_eq!(result.data_buffers().len(), num_variadic_buffers);
+    assert_eq!(result.len(), num_variadic_buffers + 1);
+
+    result
+}