Repository: apache/arrow-rs
Pull request #6380 (2024-09-11 11:06:01)

Remove "NOT YET FULLY SUPPORTED" comment from DataType::Utf8View/BinaryView
# Which issue does this PR close?

Part of #6163 

# Rationale for this change
 
As I was reviewing https://github.com/apache/arrow-rs/pull/6368 I came across a warning about Utf8View being not completely supported which was added in #5470 by @XiangpengHao when we began the implementation

Given these types are now supported across most of this crate I think this warning is no longer relevant

# What changes are included in this PR?

Remove warning in comments

# Are there any user-facing changes?
Doc changes, no functional changes

<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->


kylebarron: Do you intend to leave in the warnings about ListView and LargeListView?

https://github.com/apache/arrow-rs/blob/704f90bbf541896387bed030e12a39be308047e8/arrow-schema/src/datatype.rs#L297-L299

https://github.com/apache/arrow-rs/blob/704f90bbf541896387bed030e12a39be308047e8/arrow-schema/src/datatype.rs#L312-L314
alamb: > Do you intend to leave in the warnings about ListView and LargeListView?
> 
> https://github.com/apache/arrow-rs/blob/704f90bbf541896387bed030e12a39be308047e8/arrow-schema/src/datatype.rs#L297-L299
> 
> https://github.com/apache/arrow-rs/blob/704f90bbf541896387bed030e12a39be308047e8/arrow-schema/src/datatype.rs#L312-L314

Unfortunately I think they are still largely unimplemented (tracked by https://github.com/apache/arrow-rs/issues/5375) 

@Kikkon  has some work here https://github.com/apache/arrow-rs/issues/5501 but needs help reviewing it

Patch:
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 1937fafe3..41edc1bb1 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
@@ -118,9 +117,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-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..3a9cf17c0 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) }
     }
@@ -590,6 +593,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-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/util/bit_mask.rs b/arrow-buffer/src/util/bit_mask.rs
index 8f81cb7d0..2074f0fab 100644
--- a/arrow-buffer/src/util/bit_mask.rs
+++ b/arrow-buffer/src/util/bit_mask.rs
@@ -17,12 +17,12 @@
 
 //! 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]`
 /// 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,35 +30,131 @@ 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)
+    }
+}
 
-    null_count as usize
+/// # 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()
+    }
+}
+
+/// # 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)]
@@ -185,4 +281,40 @@ mod tests {
         assert_eq!(destination, expected_data);
         assert_eq!(result, expected_null_count);
     }
+
+    #[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 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
+            ]
+        );
+
+        // 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/display.rs b/arrow-cast/src/display.rs
index 312e79739..6373cf678 100644
--- a/arrow-cast/src/display.rs
+++ b/arrow-cast/src/display.rs
@@ -421,9 +421,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-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-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/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-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/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/lib.rs b/arrow/src/lib.rs
index 8796caf43..581f87121 100644
--- a/arrow/src/lib.rs
+++ b/arrow/src/lib.rs
@@ -361,7 +361,6 @@
 //! [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)]
@@ -370,6 +369,9 @@ 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")]
diff --git a/arrow/src/pyarrow.rs b/arrow/src/pyarrow.rs
index 336398cbf..a7b593799 100644
--- a/arrow/src/pyarrow.rs
+++ b/arrow/src/pyarrow.rs
@@ -354,7 +354,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(
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/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)


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
+}