Repository: apache/arrow-rs
Pull request #6662 (2024-10-31 15:25:09)

Fix string view LIKE checks with NULL values
# Which issue does this PR close?

- for https://github.com/apache/datafusion/issues/12637#issuecomment-2449571495

# Rationale for this change
 
Fix correctness

# What changes are included in this PR?

Fix `StringView LIKE ..` result when tested values include nulls.

# Are there any user-facing changes?

Yes

findepi: cc @goldmedal 

Patch:
diff --git a/.github/workflows/object_store.yml b/.github/workflows/object_store.yml
index bdbfc0bec..1857b3303 100644
--- a/.github/workflows/object_store.yml
+++ b/.github/workflows/object_store.yml
@@ -138,7 +138,7 @@ jobs:
 
       - name: Setup LocalStack (AWS emulation)
         run: |
-          echo "LOCALSTACK_CONTAINER=$(docker run -d -p 4566:4566 localstack/localstack:3.3.0)" >> $GITHUB_ENV
+          echo "LOCALSTACK_CONTAINER=$(docker run -d -p 4566:4566 localstack/localstack:3.8.1)" >> $GITHUB_ENV
           echo "EC2_METADATA_CONTAINER=$(docker run -d -p 1338:1338 amazon/amazon-ec2-metadata-mock:v1.9.2 --imdsv2)" >> $GITHUB_ENV
           aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket
           aws --endpoint-url=http://localhost:4566 dynamodb create-table --table-name test-table --key-schema AttributeName=path,KeyType=HASH AttributeName=etag,KeyType=RANGE --attribute-definitions AttributeName=path,AttributeType=S AttributeName=etag,AttributeType=S --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5
@@ -161,6 +161,12 @@ jobs:
       - name: Run object_store tests
         run: cargo test --features=aws,azure,gcp,http
 
+      - name: Run object_store tests (AWS native conditional put)
+        run: cargo test --features=aws
+        env:
+          AWS_CONDITIONAL_PUT: etag-put-if-not-exists
+          AWS_COPY_IF_NOT_EXISTS: multipart
+
       - name: GCS Output
         if: ${{ !cancelled() }}
         run: docker logs $GCS_CONTAINER
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index 4c967df03..6eae8e246 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -48,7 +48,7 @@ chrono = { workspace = true }
 chrono-tz = { version = "0.10", optional = true }
 num = { version = "0.4.1", default-features = false, features = ["std"] }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
-hashbrown = { version = "0.14.2", default-features = false }
+hashbrown = { version = "0.15.1", default-features = false }
 
 [features]
 ffi = ["arrow-schema/ffi", "arrow-data/ffi"]
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index 5b313913a..a35df0d20 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -33,14 +33,18 @@ use std::sync::Arc;
 
 use super::ByteArrayType;
 
-/// [Variable-size Binary View Layout]: An array of variable length bytes view arrays.
+/// [Variable-size Binary View Layout]: An array of variable length bytes views.
+///
+/// This array type is used to store variable length byte data (e.g. Strings, Binary)
+/// and has efficient operations such as `take`, `filter`, and comparison.
 ///
 /// [Variable-size Binary View Layout]: https://arrow.apache.org/docs/format/Columnar.html#variable-size-binary-view-layout
 ///
-/// This is different from [`GenericByteArray`] as it stores both an offset and
-/// length meaning that take / filter operations can be implemented without
-/// copying the underlying data. In addition, it stores an inlined prefix which
-/// can be used to speed up comparisons.
+/// This is different from [`GenericByteArray`], which also stores variable
+/// length byte data, as it represents strings with an offset and length. `take`
+/// and `filter` like operations are implemented by manipulating the "views"
+/// (`u128`) without modifying the bytes. Each view also stores an inlined
+/// prefix which speed up comparisons.
 ///
 /// # See Also
 ///
@@ -50,12 +54,6 @@ use super::ByteArrayType;
 ///
 /// [`ByteView`]: arrow_data::ByteView
 ///
-/// # Notes
-///
-/// Comparing two `GenericByteViewArray` using PartialEq compares by structure,
-/// not by value. as there are many different buffer layouts to represent the
-/// same data (e.g. different offsets, different buffer sizes, etc).
-///
 /// # Layout: "views" and buffers
 ///
 /// A `GenericByteViewArray` stores variable length byte strings. An array of
@@ -86,6 +84,52 @@ use super::ByteArrayType;
 ///   view and the entire string is stored in one of the buffers. See [`ByteView`]
 ///   to access the fields of the these views.
 ///
+/// As with other arrays, the optimized kernels in [`arrow_compute`] are likely
+/// the easiest and fastest way to work with this data. However, it is possible
+/// to access the views and buffers directly for more control.
+///
+/// For example
+///
+/// ```rust
+/// # use arrow_array::StringViewArray;
+/// # use arrow_array::Array;
+/// use arrow_data::ByteView;
+/// let array = StringViewArray::from(vec![
+///   "hello",
+///   "this string is longer than 12 bytes",
+///   "this string is also longer than 12 bytes"
+/// ]);
+///
+/// // ** Examine the first view (short string) **
+/// assert!(array.is_valid(0)); // Check for nulls
+/// let short_view: u128 = array.views()[0]; // "hello"
+/// // get length of the string
+/// let len = short_view as u32;
+/// assert_eq!(len, 5); // strings less than 12 bytes are stored in the view
+/// // SAFETY: `view` is a valid view
+/// let value = unsafe {
+///   StringViewArray::inline_value(&short_view, len as usize)
+/// };
+/// assert_eq!(value, b"hello");
+///
+/// // ** Examine the third view (long string) **
+/// assert!(array.is_valid(12)); // Check for nulls
+/// let long_view: u128 = array.views()[2]; // "this string is also longer than 12 bytes"
+/// let len = long_view as u32;
+/// assert_eq!(len, 40); // strings longer than 12 bytes are stored in the buffer
+/// let view = ByteView::from(long_view); // use ByteView to access the fields
+/// assert_eq!(view.length, 40);
+/// assert_eq!(view.buffer_index, 0);
+/// assert_eq!(view.offset, 35); // data starts after the first long string
+/// // Views for long strings store a 4 byte prefix
+/// let prefix = view.prefix.to_le_bytes();
+/// assert_eq!(&prefix, b"this");
+/// let value = array.value(2); // get the string value (see `value` implementation for how to access the bytes directly)
+/// assert_eq!(value, "this string is also longer than 12 bytes");
+/// ```
+///
+/// [`arrow_compute`]: https://docs.rs/arrow/latest/arrow/compute/index.html
+///
 /// Unlike [`GenericByteArray`], there are no constraints on the offsets other
 /// than they must point into a valid buffer. However, they can be out of order,
 /// non continuous and overlapping.
@@ -135,16 +179,6 @@ impl<T: ByteViewType + ?Sized> Clone for GenericByteViewArray<T> {
     }
 }
 
-// PartialEq
-impl<T: ByteViewType + ?Sized> PartialEq for GenericByteViewArray<T> {
-    fn eq(&self, other: &Self) -> bool {
-        other.data_type.eq(&self.data_type)
-            && other.views.eq(&self.views)
-            && other.buffers.eq(&self.buffers)
-            && other.nulls.eq(&self.nulls)
-    }
-}
-
 impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
     /// Create a new [`GenericByteViewArray`] from the provided parts, panicking on failure
     ///
@@ -694,6 +728,8 @@ where
 
 /// A [`GenericByteViewArray`] of `[u8]`
 ///
+/// See [`GenericByteViewArray`] for format and layout details.
+///
 /// # Example
 /// ```
 /// use arrow_array::BinaryViewArray;
@@ -733,6 +769,8 @@ impl From<Vec<Option<&[u8]>>> for BinaryViewArray {
 
 /// A [`GenericByteViewArray`] that stores utf8 data
 ///
+/// See [`GenericByteViewArray`] for format and layout details.
+///
 /// # Example
 /// ```
 /// use arrow_array::StringViewArray;
@@ -1004,6 +1042,6 @@ mod tests {
         };
         assert_eq!(array1, array1.clone());
         assert_eq!(array2, array2.clone());
-        assert_ne!(array1, array2);
+        assert_eq!(array1, array2);
     }
 }
diff --git a/arrow-array/src/array/mod.rs b/arrow-array/src/array/mod.rs
index 7ca59680a..4a9e54a60 100644
--- a/arrow-array/src/array/mod.rs
+++ b/arrow-array/src/array/mod.rs
@@ -654,6 +654,12 @@ impl PartialEq for StructArray {
     }
 }
 
+impl<T: ByteViewType + ?Sized> PartialEq for GenericByteViewArray<T> {
+    fn eq(&self, other: &Self) -> bool {
+        self.to_data().eq(&other.to_data())
+    }
+}
+
 /// Constructs an array using the input `data`.
 /// Returns a reference-counted `Array` instance.
 pub fn make_array(data: ArrayData) -> ArrayRef {
diff --git a/arrow-array/src/array/struct_array.rs b/arrow-array/src/array/struct_array.rs
index 059bc0b5e..d8bb1ace7 100644
--- a/arrow-array/src/array/struct_array.rs
+++ b/arrow-array/src/array/struct_array.rs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use crate::array::print_long_array;
 use crate::{make_array, new_null_array, Array, ArrayRef, RecordBatch};
 use arrow_buffer::{BooleanBuffer, Buffer, NullBuffer};
 use arrow_data::{ArrayData, ArrayDataBuilder};
@@ -404,7 +405,11 @@ impl From<Vec<(FieldRef, ArrayRef)>> for StructArray {
 
 impl std::fmt::Debug for StructArray {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(f, "StructArray\n[\n")?;
+        writeln!(f, "StructArray")?;
+        writeln!(f, "-- validity: ")?;
+        writeln!(f, "[")?;
+        print_long_array(self, f, |_array, _index, f| write!(f, "valid"))?;
+        writeln!(f, "]\n[")?;
         for (child_index, name) in self.column_names().iter().enumerate() {
             let column = self.column(child_index);
             writeln!(
@@ -731,4 +736,16 @@ mod tests {
             Arc::new(Int32Array::from(vec![Some(42), None, Some(19)])) as ArrayRef,
         )]));
     }
+
+    #[test]
+    fn test_struct_array_fmt_debug() {
+        let arr: StructArray = StructArray::new(
+            vec![Arc::new(Field::new("c", DataType::Int32, true))].into(),
+            vec![Arc::new(Int32Array::from((0..30).collect::<Vec<_>>())) as ArrayRef],
+            Some(NullBuffer::new(BooleanBuffer::from(
+                (0..30).map(|i| i % 2 == 0).collect::<Vec<_>>(),
+            ))),
+        );
+        assert_eq!(format!("{arr:?}"), "StructArray\n-- validity: \n[\n  valid,\n  null,\n  valid,\n  null,\n  valid,\n  null,\n  valid,\n  null,\n  valid,\n  null,\n  ...10 elements...,\n  valid,\n  null,\n  valid,\n  null,\n  valid,\n  null,\n  valid,\n  null,\n  valid,\n  null,\n]\n[\n-- child 0: \"c\" (Int32)\nPrimitiveArray<Int32>\n[\n  0,\n  1,\n  2,\n  3,\n  4,\n  5,\n  6,\n  7,\n  8,\n  9,\n  ...10 elements...,\n  20,\n  21,\n  22,\n  23,\n  24,\n  25,\n  26,\n  27,\n  28,\n  29,\n]\n]")
+    }
 }
diff --git a/arrow-array/src/cast.rs b/arrow-array/src/cast.rs
index 8c7cc2a12..232b29560 100644
--- a/arrow-array/src/cast.rs
+++ b/arrow-array/src/cast.rs
@@ -21,6 +21,14 @@ use crate::array::*;
 use crate::types::*;
 use arrow_data::ArrayData;
 
+/// Re-export symbols needed for downcast macros
+///
+/// Name follows `serde` convention
+#[doc(hidden)]
+pub mod __private {
+    pub use arrow_schema::{DataType, IntervalUnit, TimeUnit};
+}
+
 /// Repeats the provided pattern based on the number of comma separated identifiers
 #[doc(hidden)]
 #[macro_export]
@@ -66,28 +74,28 @@ macro_rules! repeat_pat {
 macro_rules! downcast_integer {
     ($($data_type:expr),+ => ($m:path $(, $args:tt)*), $($p:pat => $fallback:expr $(,)*)*) => {
         match ($($data_type),+) {
-            $crate::repeat_pat!(arrow_schema::DataType::Int8, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int8, $($data_type),+) => {
                 $m!($crate::types::Int8Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Int16, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int16, $($data_type),+) => {
                 $m!($crate::types::Int16Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Int32, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int32, $($data_type),+) => {
                 $m!($crate::types::Int32Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Int64, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int64, $($data_type),+) => {
                 $m!($crate::types::Int64Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::UInt8, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::UInt8, $($data_type),+) => {
                 $m!($crate::types::UInt8Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::UInt16, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::UInt16, $($data_type),+) => {
                 $m!($crate::types::UInt16Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::UInt32, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::UInt32, $($data_type),+) => {
                 $m!($crate::types::UInt32Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::UInt64, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::UInt64, $($data_type),+) => {
                 $m!($crate::types::UInt64Type $(, $args)*)
             }
             $($p => $fallback,)*
@@ -129,13 +137,13 @@ macro_rules! downcast_integer {
 macro_rules! downcast_run_end_index {
     ($($data_type:expr),+ => ($m:path $(, $args:tt)*), $($p:pat => $fallback:expr $(,)*)*) => {
         match ($($data_type),+) {
-            $crate::repeat_pat!(arrow_schema::DataType::Int16, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int16, $($data_type),+) => {
                 $m!($crate::types::Int16Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Int32, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int32, $($data_type),+) => {
                 $m!($crate::types::Int32Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Int64, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Int64, $($data_type),+) => {
                 $m!($crate::types::Int64Type $(, $args)*)
             }
             $($p => $fallback,)*
@@ -172,34 +180,34 @@ macro_rules! downcast_run_end_index {
 macro_rules! downcast_temporal {
     ($($data_type:expr),+ => ($m:path $(, $args:tt)*), $($p:pat => $fallback:expr $(,)*)*) => {
         match ($($data_type),+) {
-            $crate::repeat_pat!(arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Second), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Time32($crate::cast::__private::TimeUnit::Second), $($data_type),+) => {
                 $m!($crate::types::Time32SecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Time32(arrow_schema::TimeUnit::Millisecond), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Time32($crate::cast::__private::TimeUnit::Millisecond), $($data_type),+) => {
                 $m!($crate::types::Time32MillisecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Microsecond), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Time64($crate::cast::__private::TimeUnit::Microsecond), $($data_type),+) => {
                 $m!($crate::types::Time64MicrosecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Time64(arrow_schema::TimeUnit::Nanosecond), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Time64($crate::cast::__private::TimeUnit::Nanosecond), $($data_type),+) => {
                 $m!($crate::types::Time64NanosecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Date32, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Date32, $($data_type),+) => {
                 $m!($crate::types::Date32Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Date64, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Date64, $($data_type),+) => {
                 $m!($crate::types::Date64Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Second, _), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Timestamp($crate::cast::__private::TimeUnit::Second, _), $($data_type),+) => {
                 $m!($crate::types::TimestampSecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Millisecond, _), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Timestamp($crate::cast::__private::TimeUnit::Millisecond, _), $($data_type),+) => {
                 $m!($crate::types::TimestampMillisecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Microsecond, _), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Timestamp($crate::cast::__private::TimeUnit::Microsecond, _), $($data_type),+) => {
                 $m!($crate::types::TimestampMicrosecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Timestamp(arrow_schema::TimeUnit::Nanosecond, _), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Timestamp($crate::cast::__private::TimeUnit::Nanosecond, _), $($data_type),+) => {
                 $m!($crate::types::TimestampNanosecondType $(, $args)*)
             }
             $($p => $fallback,)*
@@ -284,40 +292,40 @@ macro_rules! downcast_primitive {
     ($($data_type:expr),+ => ($m:path $(, $args:tt)*), $($p:pat => $fallback:expr $(,)*)*) => {
         $crate::downcast_integer! {
             $($data_type),+ => ($m $(, $args)*),
-            $crate::repeat_pat!(arrow_schema::DataType::Float16, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Float16, $($data_type),+) => {
                 $m!($crate::types::Float16Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Float32, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Float32, $($data_type),+) => {
                 $m!($crate::types::Float32Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Float64, $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Float64, $($data_type),+) => {
                 $m!($crate::types::Float64Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Decimal128(_, _), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Decimal128(_, _), $($data_type),+) => {
                 $m!($crate::types::Decimal128Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Decimal256(_, _), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Decimal256(_, _), $($data_type),+) => {
                 $m!($crate::types::Decimal256Type $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Interval(arrow_schema::IntervalUnit::YearMonth), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Interval($crate::cast::__private::IntervalUnit::YearMonth), $($data_type),+) => {
                 $m!($crate::types::IntervalYearMonthType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Interval(arrow_schema::IntervalUnit::DayTime), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Interval($crate::cast::__private::IntervalUnit::DayTime), $($data_type),+) => {
                 $m!($crate::types::IntervalDayTimeType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Interval(arrow_schema::IntervalUnit::MonthDayNano), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Interval($crate::cast::__private::IntervalUnit::MonthDayNano), $($data_type),+) => {
                 $m!($crate::types::IntervalMonthDayNanoType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Duration(arrow_schema::TimeUnit::Second), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Duration($crate::cast::__private::TimeUnit::Second), $($data_type),+) => {
                 $m!($crate::types::DurationSecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Duration(arrow_schema::TimeUnit::Millisecond), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Duration($crate::cast::__private::TimeUnit::Millisecond), $($data_type),+) => {
                 $m!($crate::types::DurationMillisecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Duration(arrow_schema::TimeUnit::Microsecond), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Duration($crate::cast::__private::TimeUnit::Microsecond), $($data_type),+) => {
                 $m!($crate::types::DurationMicrosecondType $(, $args)*)
             }
-            $crate::repeat_pat!(arrow_schema::DataType::Duration(arrow_schema::TimeUnit::Nanosecond), $($data_type),+) => {
+            $crate::repeat_pat!($crate::cast::__private::DataType::Duration($crate::cast::__private::TimeUnit::Nanosecond), $($data_type),+) => {
                 $m!($crate::types::DurationNanosecondType $(, $args)*)
             }
             _ => {
@@ -462,7 +470,7 @@ macro_rules! downcast_dictionary_array {
 
     ($values:ident => $e:block $($p:pat => $fallback:expr $(,)*)*) => {
         match $values.data_type() {
-            arrow_schema::DataType::Dictionary(k, _) => {
+            $crate::cast::__private::DataType::Dictionary(k, _) => {
                 $crate::downcast_integer! {
                     k.as_ref() => ($crate::downcast_dictionary_array_helper, $values, $e),
                     k => unreachable!("unsupported dictionary key type: {}", k)
@@ -564,7 +572,7 @@ macro_rules! downcast_run_array {
 
     ($values:ident => $e:block $($p:pat => $fallback:expr $(,)*)*) => {
         match $values.data_type() {
-            arrow_schema::DataType::RunEndEncoded(k, _) => {
+            $crate::cast::__private::DataType::RunEndEncoded(k, _) => {
                 $crate::downcast_run_end_index! {
                     k.data_type() => ($crate::downcast_run_array_helper, $values, $e),
                     k => unreachable!("unsupported run end index type: {}", k)
@@ -778,19 +786,14 @@ pub trait AsArray: private::Sealed {
         self.as_bytes_opt().expect("binary array")
     }
 
-    /// Downcast this to a [`StringViewArray`] returning `None` if not possible
-    fn as_string_view(&self) -> &StringViewArray {
-        self.as_byte_view_opt().expect("string view array")
-    }
-
     /// Downcast this to a [`StringViewArray`] returning `None` if not possible
     fn as_string_view_opt(&self) -> Option<&StringViewArray> {
         self.as_byte_view_opt()
     }
 
-    /// Downcast this to a [`StringViewArray`] returning `None` if not possible
-    fn as_binary_view(&self) -> &BinaryViewArray {
-        self.as_byte_view_opt().expect("binary view array")
+    /// Downcast this to a [`StringViewArray`] panicking if not possible
+    fn as_string_view(&self) -> &StringViewArray {
+        self.as_byte_view_opt().expect("string view array")
     }
 
     /// Downcast this to a [`BinaryViewArray`] returning `None` if not possible
@@ -798,14 +801,19 @@ pub trait AsArray: private::Sealed {
         self.as_byte_view_opt()
     }
 
-    /// Downcast this to a [`GenericByteViewArray`] returning `None` if not possible
-    fn as_byte_view<T: ByteViewType>(&self) -> &GenericByteViewArray<T> {
-        self.as_byte_view_opt().expect("byte view array")
+    /// Downcast this to a [`BinaryViewArray`] panicking if not possible
+    fn as_binary_view(&self) -> &BinaryViewArray {
+        self.as_byte_view_opt().expect("binary view array")
     }
 
     /// Downcast this to a [`GenericByteViewArray`] returning `None` if not possible
     fn as_byte_view_opt<T: ByteViewType>(&self) -> Option<&GenericByteViewArray<T>>;
 
+    /// Downcast this to a [`GenericByteViewArray`] panicking if not possible
+    fn as_byte_view<T: ByteViewType>(&self) -> &GenericByteViewArray<T> {
+        self.as_byte_view_opt().expect("byte view array")
+    }
+
     /// Downcast this to a [`StructArray`] returning `None` if not possible
     fn as_struct_opt(&self) -> Option<&StructArray>;
 
diff --git a/arrow-array/src/ffi.rs b/arrow-array/src/ffi.rs
index 29414eae6..4426e0986 100644
--- a/arrow-array/src/ffi.rs
+++ b/arrow-array/src/ffi.rs
@@ -302,8 +302,8 @@ impl ImportedArrowArray<'_> {
         let len = self.array.len();
         let offset = self.array.offset();
         let null_count = match &self.data_type {
-            DataType::Null => 0,
-            _ => self.array.null_count(),
+            DataType::Null => Some(0),
+            _ => self.array.null_count_opt(),
         };
 
         let data_layout = layout(&self.data_type);
@@ -329,7 +329,7 @@ impl ImportedArrowArray<'_> {
             ArrayData::new_unchecked(
                 self.data_type,
                 len,
-                Some(null_count),
+                null_count,
                 null_bit_buffer,
                 offset,
                 buffers,
@@ -635,6 +635,38 @@ mod tests_to_then_from_ffi {
     }
     // case with nulls is tested in the docs, through the example on this module.
 
+    #[test]
+    fn test_null_count_handling() {
+        let int32_data = ArrayData::builder(DataType::Int32)
+            .len(10)
+            .add_buffer(Buffer::from_slice_ref([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]))
+            .null_bit_buffer(Some(Buffer::from([0b01011111, 0b00000001])))
+            .build()
+            .unwrap();
+        let mut ffi_array = FFI_ArrowArray::new(&int32_data);
+        assert_eq!(3, ffi_array.null_count());
+        assert_eq!(Some(3), ffi_array.null_count_opt());
+        // Simulating uninitialized state
+        unsafe {
+            ffi_array.set_null_count(-1);
+        }
+        assert_eq!(None, ffi_array.null_count_opt());
+        let int32_data = unsafe { from_ffi_and_data_type(ffi_array, DataType::Int32) }.unwrap();
+        assert_eq!(3, int32_data.null_count());
+
+        let null_data = &ArrayData::new_null(&DataType::Null, 10);
+        let mut ffi_array = FFI_ArrowArray::new(null_data);
+        assert_eq!(10, ffi_array.null_count());
+        assert_eq!(Some(10), ffi_array.null_count_opt());
+        // Simulating uninitialized state
+        unsafe {
+            ffi_array.set_null_count(-1);
+        }
+        assert_eq!(None, ffi_array.null_count_opt());
+        let null_data = unsafe { from_ffi_and_data_type(ffi_array, DataType::Null) }.unwrap();
+        assert_eq!(0, null_data.null_count());
+    }
+
     fn test_generic_string<Offset: OffsetSizeTrait>() -> Result<()> {
         // create an array natively
         let array = GenericStringArray::<Offset>::from(vec![Some("a"), None, Some("aaa")]);
diff --git a/arrow-data/src/ffi.rs b/arrow-data/src/ffi.rs
index cd283d326..04599bc47 100644
--- a/arrow-data/src/ffi.rs
+++ b/arrow-data/src/ffi.rs
@@ -271,6 +271,21 @@ impl FFI_ArrowArray {
         self.null_count as usize
     }
 
+    /// Returns the null count, checking for validity
+    #[inline]
+    pub fn null_count_opt(&self) -> Option<usize> {
+        usize::try_from(self.null_count).ok()
+    }
+
+    /// Set the null count of the array
+    ///
+    /// # Safety
+    /// Null count must match that of null buffer
+    #[inline]
+    pub unsafe fn set_null_count(&mut self, null_count: i64) {
+        self.null_count = null_count;
+    }
+
     /// Returns the buffer at the provided index
     ///
     /// # Panic
diff --git a/arrow-data/src/transform/variable_size.rs b/arrow-data/src/transform/variable_size.rs
index fa1592d97..ec0174bf8 100644
--- a/arrow-data/src/transform/variable_size.rs
+++ b/arrow-data/src/transform/variable_size.rs
@@ -34,7 +34,7 @@ fn extend_offset_values<T: ArrowNativeType + AsPrimitive<usize>>(
     len: usize,
 ) {
     let start_values = offsets[start].as_();
-    let end_values = offsets[start + len].as_();
+    let end_values: usize = offsets[start + len].as_();
     let new_values = &values[start_values..end_values];
     buffer.extend_from_slice(new_values);
 }
diff --git a/arrow-flight/src/encode.rs b/arrow-flight/src/encode.rs
index acfbd9b53..ec4fe323b 100644
--- a/arrow-flight/src/encode.rs
+++ b/arrow-flight/src/encode.rs
@@ -104,6 +104,41 @@ use futures::{ready, stream::BoxStream, Stream, StreamExt};
 /// # }
 /// ```
 ///
+/// # Example: Determining schema of encoded data
+///
+/// Encoding flight data may hydrate dictionaries, see [`DictionaryHandling`] for more information,
+/// which changes the schema of the encoded data compared to the input record batches.
+/// The fully hydrated schema can be accessed using the [`FlightDataEncoder::known_schema`] method
+/// and explicitly informing the builder of the schema using [`FlightDataEncoderBuilder::with_schema`].
+///
+/// ```
+/// # use std::sync::Arc;
+/// # use arrow_array::{ArrayRef, RecordBatch, UInt32Array};
+/// # async fn f() {
+/// # let c1 = UInt32Array::from(vec![1, 2, 3, 4, 5, 6]);
+/// # let batch = RecordBatch::try_from_iter(vec![
+/// #      ("a", Arc::new(c1) as ArrayRef)
+/// #   ])
+/// #   .expect("cannot create record batch");
+/// use arrow_flight::encode::FlightDataEncoderBuilder;
+///
+/// // Get the schema of the input stream
+/// let schema = batch.schema();
+///
+/// // Get an input stream of Result<RecordBatch, FlightError>
+/// let input_stream = futures::stream::iter(vec![Ok(batch)]);
+///
+/// // Build a stream of `Result<FlightData>` (e.g. to return for do_get)
+/// let flight_data_stream = FlightDataEncoderBuilder::new()
+///  // Inform the builder of the input stream schema
+///  .with_schema(schema)
+///  .build(input_stream);
+///
+/// // Retrieve the schema of the encoded data
+/// let encoded_schema = flight_data_stream.known_schema();
+/// # }
+/// ```
+///
 /// [`FlightService::do_get`]: crate::flight_service_server::FlightService::do_get
 /// [`FlightError`]: crate::error::FlightError
 #[derive(Debug)]
@@ -287,6 +322,12 @@ impl FlightDataEncoder {
         encoder
     }
 
+    /// Report the schema of the encoded data when known.
+    /// A schema is known when provided via the [`FlightDataEncoderBuilder::with_schema`] method.
+    pub fn known_schema(&self) -> Option<SchemaRef> {
+        self.schema.clone()
+    }
+
     /// Place the `FlightData` in the queue to send
     fn queue_message(&mut self, mut data: FlightData) {
         if let Some(descriptor) = self.descriptor.take() {
@@ -792,6 +833,53 @@ mod tests {
         verify_flight_round_trip(vec![batch1, batch2]).await;
     }
 
+    #[tokio::test]
+    async fn test_dictionary_hydration_known_schema() {
+        let arr1: DictionaryArray<UInt16Type> = vec!["a", "a", "b"].into_iter().collect();
+        let arr2: DictionaryArray<UInt16Type> = vec!["c", "c", "d"].into_iter().collect();
+
+        let schema = Arc::new(Schema::new(vec![Field::new_dictionary(
+            "dict",
+            DataType::UInt16,
+            DataType::Utf8,
+            false,
+        )]));
+        let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr2)]).unwrap();
+
+        let stream = futures::stream::iter(vec![Ok(batch1), Ok(batch2)]);
+
+        let encoder = FlightDataEncoderBuilder::default()
+            .with_schema(schema)
+            .build(stream);
+        let expected_schema =
+            Arc::new(Schema::new(vec![Field::new("dict", DataType::Utf8, false)]));
+        assert_eq!(Some(expected_schema), encoder.known_schema())
+    }
+
+    #[tokio::test]
+    async fn test_dictionary_resend_known_schema() {
+        let arr1: DictionaryArray<UInt16Type> = vec!["a", "a", "b"].into_iter().collect();
+        let arr2: DictionaryArray<UInt16Type> = vec!["c", "c", "d"].into_iter().collect();
+
+        let schema = Arc::new(Schema::new(vec![Field::new_dictionary(
+            "dict",
+            DataType::UInt16,
+            DataType::Utf8,
+            false,
+        )]));
+        let batch1 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr1)]).unwrap();
+        let batch2 = RecordBatch::try_new(schema.clone(), vec![Arc::new(arr2)]).unwrap();
+
+        let stream = futures::stream::iter(vec![Ok(batch1), Ok(batch2)]);
+
+        let encoder = FlightDataEncoderBuilder::default()
+            .with_dictionary_handling(DictionaryHandling::Resend)
+            .with_schema(schema.clone())
+            .build(stream);
+        assert_eq!(Some(schema), encoder.known_schema())
+    }
+
     #[tokio::test]
     async fn test_multiple_dictionaries_resend() {
         // Create a schema with two dictionary fields that have the same dict ID
diff --git a/arrow-json/src/writer/encoder.rs b/arrow-json/src/writer/encoder.rs
index 84ed384cf..ed430fe6a 100644
--- a/arrow-json/src/writer/encoder.rs
+++ b/arrow-json/src/writer/encoder.rs
@@ -454,7 +454,7 @@ impl Encoder for ArrayFormatter<'_> {
 /// A newtype wrapper around [`ArrayFormatter`] that skips surrounding the value with `"`
 struct RawArrayFormatter<'a>(ArrayFormatter<'a>);
 
-impl<'a> Encoder for RawArrayFormatter<'a> {
+impl Encoder for RawArrayFormatter<'_> {
     fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
         let _ = write!(out, "{}", self.0.value(idx));
     }
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index d616b01e7..451b04485 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -436,18 +436,21 @@ where
 
     let mut start = 0i64;
     let mut i = 0;
-    let filter_values = pred.filter.values();
     let mut count = R::default_value();
+    let filter_values = pred.filter.values();
 
     for end in run_ends.inner().into_iter().map(|i| (*i).into()) {
         let mut keep = false;
-        // in filter_array the predicate array is checked to have the same len as the run end array
-        // this means the largest value in the run_ends is == to pred.len()
-        // so we're always within bounds when calling value_unchecked
-        for pred in (start..end).map(|i| unsafe { filter_values.value_unchecked(i as usize) }) {
+
+        for pred in filter_values
+            .iter()
+            .skip(start as usize)
+            .take((end - start) as usize)
+        {
             count += R::Native::from(pred);
             keep |= pred
         }
+
         // this is to avoid branching
         new_run_ends[i] = count;
         i += keep as usize;
@@ -579,7 +582,6 @@ fn filter_native<T: ArrowNativeType>(values: &[T], predicate: &FilterPredicate)
         }
         IterationStrategy::Indices(indices) => {
             let iter = indices.iter().map(|x| values[*x]);
-
             // SAFETY: `Vec::iter` is trusted length
             unsafe { MutableBuffer::from_trusted_len_iter(iter) }
         }
@@ -615,8 +617,8 @@ where
 struct FilterBytes<'a, OffsetSize> {
     src_offsets: &'a [OffsetSize],
     src_values: &'a [u8],
-    dst_offsets: MutableBuffer,
-    dst_values: MutableBuffer,
+    dst_offsets: Vec<OffsetSize>,
+    dst_values: Vec<u8>,
     cur_offset: OffsetSize,
 }
 
@@ -628,10 +630,10 @@ where
     where
         T: ByteArrayType<Offset = OffsetSize>,
     {
-        let num_offsets_bytes = (capacity + 1) * std::mem::size_of::<OffsetSize>();
-        let mut dst_offsets = MutableBuffer::new(num_offsets_bytes);
-        let dst_values = MutableBuffer::new(0);
+        let dst_values = Vec::new();
+        let mut dst_offsets: Vec<OffsetSize> = Vec::with_capacity(capacity + 1);
         let cur_offset = OffsetSize::from_usize(0).unwrap();
+
         dst_offsets.push(cur_offset);
 
         Self {
@@ -661,13 +663,15 @@ where
 
     /// Extends the in-progress array by the indexes in the provided iterator
     fn extend_idx(&mut self, iter: impl Iterator<Item = usize>) {
-        for idx in iter {
-            let (start, end, len) = self.get_value_range(idx);
+        self.dst_offsets.extend(iter.map(|idx| {
+            let start = self.src_offsets[idx].as_usize();
+            let end = self.src_offsets[idx + 1].as_usize();
+            let len = OffsetSize::from_usize(end - start).expect("illegal offset range");
             self.cur_offset += len;
-            self.dst_offsets.push(self.cur_offset);
             self.dst_values
                 .extend_from_slice(&self.src_values[start..end]);
-        }
+            self.cur_offset
+        }));
     }
 
     /// Extends the in-progress array by the ranges in the provided iterator
@@ -1280,6 +1284,26 @@ mod tests {
         assert_eq!(0, actual.len());
     }
 
+    #[test]
+    fn test_filter_run_end_encoding_array_max_value_gt_predicate_len() {
+        let run_ends = Int64Array::from(vec![2, 3, 8, 10]);
+        let values = Int64Array::from(vec![7, -2, 9, -8]);
+        let a = RunArray::try_new(&run_ends, &values).expect("Failed to create RunArray");
+        let b = BooleanArray::from(vec![false, true, true]);
+        let c = filter(&a, &b).unwrap();
+        let actual: &RunArray<Int64Type> = as_run_array(&c);
+        assert_eq!(2, actual.len());
+
+        let expected = RunArray::try_new(
+            &Int64Array::from(vec![1, 2]),
+            &Int64Array::from(vec![7, -2]),
+        )
+        .expect("Failed to make expected RunArray test is broken");
+
+        assert_eq!(&actual.run_ends().values(), &expected.run_ends().values());
+        assert_eq!(actual.values(), expected.values())
+    }
+
     #[test]
     fn test_filter_dictionary_array() {
         let values = [Some("hello"), None, Some("world"), Some("!")];
diff --git a/arrow-string/src/length.rs b/arrow-string/src/length.rs
index 97f876a9f..6a28d44ea 100644
--- a/arrow-string/src/length.rs
+++ b/arrow-string/src/length.rs
@@ -137,6 +137,15 @@ pub fn bit_length(array: &dyn Array) -> Result<ArrayRef, ArrowError> {
             let list = array.as_string::<i64>();
             Ok(bit_length_impl::<Int64Type>(list.offsets(), list.nulls()))
         }
+        DataType::Utf8View => {
+            let list = array.as_string_view();
+            let values = list
+                .views()
+                .iter()
+                .map(|view| (*view as i32).wrapping_mul(8))
+                .collect();
+            Ok(Arc::new(Int32Array::new(values, array.nulls().cloned())))
+        }
         DataType::Binary => {
             let list = array.as_binary::<i32>();
             Ok(bit_length_impl::<Int32Type>(list.offsets(), list.nulls()))
@@ -462,6 +471,35 @@ mod tests {
             })
     }
 
+    #[test]
+    fn bit_length_test_utf8view() {
+        bit_length_cases()
+            .into_iter()
+            .for_each(|(input, len, expected)| {
+                let string_array = StringViewArray::from(input);
+                let result = bit_length(&string_array).unwrap();
+                assert_eq!(len, result.len());
+                let result = result.as_any().downcast_ref::<Int32Array>().unwrap();
+                expected.iter().enumerate().for_each(|(i, value)| {
+                    assert_eq!(*value, result.value(i));
+                });
+            })
+    }
+
+    #[test]
+    fn bit_length_null_utf8view() {
+        bit_length_null_cases()
+            .into_iter()
+            .for_each(|(input, len, expected)| {
+                let array = StringArray::from(input);
+                let result = bit_length(&array).unwrap();
+                assert_eq!(len, result.len());
+                let result = result.as_any().downcast_ref::<Int32Array>().unwrap();
+
+                let expected: Int32Array = expected.into();
+                assert_eq!(&expected, result);
+            })
+    }
     #[test]
     fn bit_length_binary() {
         let value: Vec<&[u8]> = vec![b"one", &[0xff, 0xf8], b"three"];
diff --git a/arrow-string/src/like.rs b/arrow-string/src/like.rs
index 5c76d5c81..fee3b792a 100644
--- a/arrow-string/src/like.rs
+++ b/arrow-string/src/like.rs
@@ -1708,7 +1708,93 @@ mod tests {
     }
 
     #[test]
-    fn like_scalar_null() {
+    fn string_null_like_pattern() {
+        // Different patterns have different execution code paths
+        for pattern in &[
+            "",           // can execute as equality check
+            "_",          // can execute as length check
+            "%",          // can execute as starts_with("") or non-null check
+            "a%",         // can execute as starts_with("a")
+            "%a",         // can execute as ends_with("")
+            "a%b",        // can execute as starts_with("a") && ends_with("b")
+            "%a%",        // can_execute as contains("a")
+            "%a%b_c_d%e", // can_execute as regular expression
+        ] {
+            let a = Scalar::new(StringArray::new_null(1));
+            let b = StringArray::new_scalar(pattern);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+
+            let a = Scalar::new(StringArray::new_null(1));
+            let b = StringArray::from_iter_values([pattern]);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+
+            let a = StringArray::new_null(1);
+            let b = StringArray::from_iter_values([pattern]);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+
+            let a = StringArray::new_null(1);
+            let b = StringArray::new_scalar(pattern);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+        }
+    }
+
+    #[test]
+    fn string_view_null_like_pattern() {
+        // Different patterns have different execution code paths
+        for pattern in &[
+            "",           // can execute as equality check
+            "_",          // can execute as length check
+            "%",          // can execute as starts_with("") or non-null check
+            "a%",         // can execute as starts_with("a")
+            "%a",         // can execute as ends_with("")
+            "a%b",        // can execute as starts_with("a") && ends_with("b")
+            "%a%",        // can_execute as contains("a")
+            "%a%b_c_d%e", // can_execute as regular expression
+        ] {
+            let a = Scalar::new(StringViewArray::new_null(1));
+            let b = StringViewArray::new_scalar(pattern);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+
+            let a = Scalar::new(StringViewArray::new_null(1));
+            let b = StringViewArray::from_iter_values([pattern]);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+
+            let a = StringViewArray::new_null(1);
+            let b = StringViewArray::from_iter_values([pattern]);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+
+            let a = StringViewArray::new_null(1);
+            let b = StringViewArray::new_scalar(pattern);
+            let r = like(&a, &b).unwrap();
+            assert_eq!(r.len(), 1, "With pattern {pattern}");
+            assert_eq!(r.null_count(), 1, "With pattern {pattern}");
+            assert!(r.is_null(0), "With pattern {pattern}");
+        }
+    }
+
+    #[test]
+    fn string_like_scalar_null() {
         let a = StringArray::new_scalar("a");
         let b = Scalar::new(StringArray::new_null(1));
         let r = like(&a, &b).unwrap();
@@ -1737,4 +1823,35 @@ mod tests {
         assert_eq!(r.null_count(), 1);
         assert!(r.is_null(0));
     }
+
+    #[test]
+    fn string_view_like_scalar_null() {
+        let a = StringViewArray::new_scalar("a");
+        let b = Scalar::new(StringViewArray::new_null(1));
+        let r = like(&a, &b).unwrap();
+        assert_eq!(r.len(), 1);
+        assert_eq!(r.null_count(), 1);
+        assert!(r.is_null(0));
+
+        let a = StringViewArray::from_iter_values(["a"]);
+        let b = Scalar::new(StringViewArray::new_null(1));
+        let r = like(&a, &b).unwrap();
+        assert_eq!(r.len(), 1);
+        assert_eq!(r.null_count(), 1);
+        assert!(r.is_null(0));
+
+        let a = StringViewArray::from_iter_values(["a"]);
+        let b = StringViewArray::new_null(1);
+        let r = like(&a, &b).unwrap();
+        assert_eq!(r.len(), 1);
+        assert_eq!(r.null_count(), 1);
+        assert!(r.is_null(0));
+
+        let a = StringViewArray::new_scalar("a");
+        let b = StringViewArray::new_null(1);
+        let r = like(&a, &b).unwrap();
+        assert_eq!(r.len(), 1);
+        assert_eq!(r.null_count(), 1);
+        assert!(r.is_null(0));
+    }
 }
diff --git a/arrow-string/src/predicate.rs b/arrow-string/src/predicate.rs
index f559088e6..1d141e6b0 100644
--- a/arrow-string/src/predicate.rs
+++ b/arrow-string/src/predicate.rs
@@ -15,7 +15,8 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use arrow_array::{ArrayAccessor, BooleanArray, StringViewArray};
+use arrow_array::{Array, ArrayAccessor, BooleanArray, StringViewArray};
+use arrow_buffer::BooleanBuffer;
 use arrow_schema::ArrowError;
 use memchr::memchr2;
 use memchr::memmem::Finder;
@@ -114,30 +115,21 @@ impl<'a> Predicate<'a> {
             Predicate::IEqAscii(v) => BooleanArray::from_unary(array, |haystack| {
                 haystack.eq_ignore_ascii_case(v) != negate
             }),
-            Predicate::Contains(finder) => {
-                if let Some(string_view_array) = array.as_any().downcast_ref::<StringViewArray>() {
-                    BooleanArray::from(
-                        string_view_array
-                            .bytes_iter()
-                            .map(|haystack| finder.find(haystack).is_some() != negate)
-                            .collect::<Vec<_>>(),
-                    )
-                } else {
-                    BooleanArray::from_unary(array, |haystack| {
-                        finder.find(haystack.as_bytes()).is_some() != negate
-                    })
-                }
-            }
+            Predicate::Contains(finder) => BooleanArray::from_unary(array, |haystack| {
+                finder.find(haystack.as_bytes()).is_some() != negate
+            }),
             Predicate::StartsWith(v) => {
                 if let Some(string_view_array) = array.as_any().downcast_ref::<StringViewArray>() {
-                    BooleanArray::from(
+                    let nulls = string_view_array.logical_nulls();
+                    let values = BooleanBuffer::from(
                         string_view_array
                             .prefix_bytes_iter(v.len())
                             .map(|haystack| {
                                 equals_bytes(haystack, v.as_bytes(), equals_kernel) != negate
                             })
                             .collect::<Vec<_>>(),
-                    )
+                    );
+                    BooleanArray::new(values, nulls)
                 } else {
                     BooleanArray::from_unary(array, |haystack| {
                         starts_with(haystack, v, equals_kernel) != negate
@@ -146,6 +138,7 @@ impl<'a> Predicate<'a> {
             }
             Predicate::IStartsWithAscii(v) => {
                 if let Some(string_view_array) = array.as_any().downcast_ref::<StringViewArray>() {
+                    // TODO respect null buffer
                     BooleanArray::from(
                         string_view_array
                             .prefix_bytes_iter(v.len())
@@ -166,14 +159,16 @@ impl<'a> Predicate<'a> {
             }
             Predicate::EndsWith(v) => {
                 if let Some(string_view_array) = array.as_any().downcast_ref::<StringViewArray>() {
-                    BooleanArray::from(
+                    let nulls = string_view_array.logical_nulls();
+                    let values = BooleanBuffer::from(
                         string_view_array
                             .suffix_bytes_iter(v.len())
                             .map(|haystack| {
                                 equals_bytes(haystack, v.as_bytes(), equals_kernel) != negate
                             })
                             .collect::<Vec<_>>(),
-                    )
+                    );
+                    BooleanArray::new(values, nulls)
                 } else {
                     BooleanArray::from_unary(array, |haystack| {
                         ends_with(haystack, v, equals_kernel) != negate
@@ -182,6 +177,7 @@ impl<'a> Predicate<'a> {
             }
             Predicate::IEndsWithAscii(v) => {
                 if let Some(string_view_array) = array.as_any().downcast_ref::<StringViewArray>() {
+                    // TODO respect null buffer
                     BooleanArray::from(
                         string_view_array
                             .suffix_bytes_iter(v.len())
@@ -239,7 +235,7 @@ fn equals_kernel((n, h): (&u8, &u8)) -> bool {
 }
 
 fn equals_ignore_ascii_case_kernel((n, h): (&u8, &u8)) -> bool {
-    n.to_ascii_lowercase() == h.to_ascii_lowercase()
+    n.eq_ignore_ascii_case(h)
 }
 
 /// Transforms a like `pattern` to a regex compatible pattern. To achieve that, it does:
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index cab2ac31c..86d1392eb 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -46,7 +46,7 @@ walkdir = "2"
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
-quick-xml = { version = "0.36.0", features = ["serialize", "overlapped-lists"], optional = true }
+quick-xml = { version = "0.37.0", features = ["serialize", "overlapped-lists"], optional = true }
 serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
 serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 4b4d0b6e3..a610e6351 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -28,8 +28,8 @@ use crate::client::header::{get_put_result, get_version};
 use crate::client::list::ListClient;
 use crate::client::retry::RetryExt;
 use crate::client::s3::{
-    CompleteMultipartUpload, CompleteMultipartUploadResult, InitiateMultipartUploadResult,
-    ListResponse,
+    CompleteMultipartUpload, CompleteMultipartUploadResult, CopyPartResult,
+    InitiateMultipartUploadResult, ListResponse,
 };
 use crate::client::GetOptionsExt;
 use crate::multipart::PartId;
@@ -98,8 +98,11 @@ pub(crate) enum Error {
     #[snafu(display("Error getting create multipart response body: {}", source))]
     CreateMultipartResponseBody { source: reqwest::Error },
 
-    #[snafu(display("Error performing complete multipart request: {}", source))]
-    CompleteMultipartRequest { source: crate::client::retry::Error },
+    #[snafu(display("Error performing complete multipart request: {}: {}", path, source))]
+    CompleteMultipartRequest {
+        source: crate::client::retry::Error,
+        path: String,
+    },
 
     #[snafu(display("Error getting complete multipart response body: {}", source))]
     CompleteMultipartResponseBody { source: reqwest::Error },
@@ -118,13 +121,32 @@ pub(crate) enum Error {
 
 impl From<Error> for crate::Error {
     fn from(err: Error) -> Self {
-        Self::Generic {
-            store: STORE,
-            source: Box::new(err),
+        match err {
+            Error::CompleteMultipartRequest { source, path } => source.error(STORE, path),
+            _ => Self::Generic {
+                store: STORE,
+                source: Box::new(err),
+            },
         }
     }
 }
 
+pub(crate) enum PutPartPayload<'a> {
+    Part(PutPayload),
+    Copy(&'a Path),
+}
+
+impl Default for PutPartPayload<'_> {
+    fn default() -> Self {
+        Self::Part(PutPayload::default())
+    }
+}
+
+pub(crate) enum CompleteMultipartMode {
+    Overwrite,
+    Create,
+}
+
 #[derive(Deserialize)]
 #[serde(rename_all = "PascalCase", rename = "DeleteResult")]
 struct BatchDeleteResponse {
@@ -605,15 +627,24 @@ impl S3Client {
         path: &Path,
         upload_id: &MultipartId,
         part_idx: usize,
-        data: PutPayload,
+        data: PutPartPayload<'_>,
     ) -> Result<PartId> {
+        let is_copy = matches!(data, PutPartPayload::Copy(_));
         let part = (part_idx + 1).to_string();
 
         let mut request = self
             .request(Method::PUT, path)
-            .with_payload(data)
             .query(&[("partNumber", &part), ("uploadId", upload_id)])
             .idempotent(true);
+
+        request = match data {
+            PutPartPayload::Part(payload) => request.with_payload(payload),
+            PutPartPayload::Copy(path) => request.header(
+                "x-amz-copy-source",
+                &format!("{}/{}", self.config.bucket, encode_path(path)),
+            ),
+        };
+
         if self
             .config
             .encryption_headers
@@ -625,21 +656,48 @@ impl S3Client {
         }
         let response = request.send().await?;
 
-        let content_id = get_etag(response.headers()).context(MetadataSnafu)?;
+        let content_id = match is_copy {
+            false => get_etag(response.headers()).context(MetadataSnafu)?,
+            true => {
+                let response = response
+                    .bytes()
+                    .await
+                    .context(CreateMultipartResponseBodySnafu)?;
+                let response: CopyPartResult = quick_xml::de::from_reader(response.reader())
+                    .context(InvalidMultipartResponseSnafu)?;
+                response.e_tag
+            }
+        };
         Ok(PartId { content_id })
     }
 
+    pub(crate) async fn abort_multipart(&self, location: &Path, upload_id: &str) -> Result<()> {
+        self.request(Method::DELETE, location)
+            .query(&[("uploadId", upload_id)])
+            .with_encryption_headers()
+            .send()
+            .await?;
+
+        Ok(())
+    }
+
     pub(crate) async fn complete_multipart(
         &self,
         location: &Path,
         upload_id: &str,
         parts: Vec<PartId>,
+        mode: CompleteMultipartMode,
     ) -> Result<PutResult> {
         let parts = if parts.is_empty() {
             // If no parts were uploaded, upload an empty part
             // otherwise the completion request will fail
             let part = self
-                .put_part(location, &upload_id.to_string(), 0, PutPayload::default())
+                .put_part(
+                    location,
+                    &upload_id.to_string(),
+                    0,
+                    PutPartPayload::default(),
+                )
                 .await?;
             vec![part]
         } else {
@@ -651,18 +709,27 @@ impl S3Client {
         let credential = self.config.get_session_credential().await?;
         let url = self.config.path_url(location);
 
-        let response = self
+        let request = self
             .client
             .request(Method::POST, url)
             .query(&[("uploadId", upload_id)])
             .body(body)
-            .with_aws_sigv4(credential.authorizer(), None)
+            .with_aws_sigv4(credential.authorizer(), None);
+
+        let request = match mode {
+            CompleteMultipartMode::Overwrite => request,
+            CompleteMultipartMode::Create => request.header("If-None-Match", "*"),
+        };
+
+        let response = request
             .retryable(&self.config.retry_config)
             .idempotent(true)
             .retry_error_body(true)
             .send()
             .await
-            .context(CompleteMultipartRequestSnafu)?;
+            .context(CompleteMultipartRequestSnafu {
+                path: location.as_ref(),
+            })?;
 
         let version = get_version(response.headers(), VERSION_HEADER).context(MetadataSnafu)?;
 
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index a27ed0533..b238d90eb 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -36,7 +36,7 @@ use reqwest::{Method, StatusCode};
 use std::{sync::Arc, time::Duration};
 use url::Url;
 
-use crate::aws::client::{RequestError, S3Client};
+use crate::aws::client::{CompleteMultipartMode, PutPartPayload, RequestError, S3Client};
 use crate::client::get::GetClientExt;
 use crate::client::list::ListClientExt;
 use crate::client::CredentialProvider;
@@ -169,7 +169,10 @@ impl ObjectStore for AmazonS3 {
         match (opts.mode, &self.client.config.conditional_put) {
             (PutMode::Overwrite, _) => request.idempotent(true).do_put().await,
             (PutMode::Create | PutMode::Update(_), None) => Err(Error::NotImplemented),
-            (PutMode::Create, Some(S3ConditionalPut::ETagMatch)) => {
+            (
+                PutMode::Create,
+                Some(S3ConditionalPut::ETagMatch | S3ConditionalPut::ETagPutIfNotExists),
+            ) => {
                 match request.header(&IF_NONE_MATCH, "*").do_put().await {
                     // Technically If-None-Match should return NotModified but some stores,
                     // such as R2, instead return PreconditionFailed
@@ -193,6 +196,7 @@ impl ObjectStore for AmazonS3 {
                     source: "ETag required for conditional put".to_string().into(),
                 })?;
                 match put {
+                    S3ConditionalPut::ETagPutIfNotExists => Err(Error::NotImplemented),
                     S3ConditionalPut::ETagMatch => {
                         request.header(&IF_MATCH, etag.as_str()).do_put().await
                     }
@@ -293,6 +297,47 @@ impl ObjectStore for AmazonS3 {
         let (k, v, status) = match &self.client.config.copy_if_not_exists {
             Some(S3CopyIfNotExists::Header(k, v)) => (k, v, StatusCode::PRECONDITION_FAILED),
             Some(S3CopyIfNotExists::HeaderWithStatus(k, v, status)) => (k, v, *status),
+            Some(S3CopyIfNotExists::Multipart) => {
+                let upload_id = self
+                    .client
+                    .create_multipart(to, PutMultipartOpts::default())
+                    .await?;
+
+                let res = async {
+                    let part_id = self
+                        .client
+                        .put_part(to, &upload_id, 0, PutPartPayload::Copy(from))
+                        .await?;
+                    match self
+                        .client
+                        .complete_multipart(
+                            to,
+                            &upload_id,
+                            vec![part_id],
+                            CompleteMultipartMode::Create,
+                        )
+                        .await
+                    {
+                        Err(e @ Error::Precondition { .. }) => Err(Error::AlreadyExists {
+                            path: to.to_string(),
+                            source: Box::new(e),
+                        }),
+                        Ok(_) => Ok(()),
+                        Err(e) => Err(e),
+                    }
+                }
+                .await;
+
+                // If the multipart upload failed, make a best effort attempt to
+                // clean it up. It's the caller's responsibility to add a
+                // lifecycle rule if guaranteed cleanup is required, as we
+                // cannot protect against an ill-timed process crash.
+                if res.is_err() {
+                    let _ = self.client.abort_multipart(to, &upload_id).await;
+                }
+
+                return res;
+            }
             Some(S3CopyIfNotExists::Dynamo(lock)) => {
                 return lock.copy_if_not_exists(&self.client, from, to).await
             }
@@ -340,7 +385,12 @@ impl MultipartUpload for S3MultiPartUpload {
         Box::pin(async move {
             let part = state
                 .client
-                .put_part(&state.location, &state.upload_id, idx, data)
+                .put_part(
+                    &state.location,
+                    &state.upload_id,
+                    idx,
+                    PutPartPayload::Part(data),
+                )
                 .await?;
             state.parts.put(idx, part);
             Ok(())
@@ -352,7 +402,12 @@ impl MultipartUpload for S3MultiPartUpload {
 
         self.state
             .client
-            .complete_multipart(&self.state.location, &self.state.upload_id, parts)
+            .complete_multipart(
+                &self.state.location,
+                &self.state.upload_id,
+                parts,
+                CompleteMultipartMode::Overwrite,
+            )
             .await
     }
 
@@ -384,7 +439,9 @@ impl MultipartStore for AmazonS3 {
         part_idx: usize,
         data: PutPayload,
     ) -> Result<PartId> {
-        self.client.put_part(path, id, part_idx, data).await
+        self.client
+            .put_part(path, id, part_idx, PutPartPayload::Part(data))
+            .await
     }
 
     async fn complete_multipart(
@@ -393,7 +450,9 @@ impl MultipartStore for AmazonS3 {
         id: &MultipartId,
         parts: Vec<PartId>,
     ) -> Result<PutResult> {
-        self.client.complete_multipart(path, id, parts).await
+        self.client
+            .complete_multipart(path, id, parts, CompleteMultipartMode::Overwrite)
+            .await
     }
 
     async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()> {
@@ -427,7 +486,6 @@ mod tests {
         let integration = config.build().unwrap();
         let config = &integration.client.config;
         let test_not_exists = config.copy_if_not_exists.is_some();
-        let test_conditional_put = config.conditional_put.is_some();
 
         put_get_delete_list(&integration).await;
         get_opts(&integration).await;
@@ -458,8 +516,9 @@ mod tests {
         if test_not_exists {
             copy_if_not_exists(&integration).await;
         }
-        if test_conditional_put {
-            put_opts(&integration, true).await;
+        if let Some(conditional_put) = &config.conditional_put {
+            let supports_update = !matches!(conditional_put, S3ConditionalPut::ETagPutIfNotExists);
+            put_opts(&integration, supports_update).await;
         }
 
         // run integration test with unsigned payload enabled
diff --git a/object_store/src/aws/precondition.rs b/object_store/src/aws/precondition.rs
index ad9e21537..e50580527 100644
--- a/object_store/src/aws/precondition.rs
+++ b/object_store/src/aws/precondition.rs
@@ -46,6 +46,21 @@ pub enum S3CopyIfNotExists {
     ///
     /// Encoded as `header-with-status:<HEADER_NAME>:<HEADER_VALUE>:<STATUS>` ignoring whitespace
     HeaderWithStatus(String, String, reqwest::StatusCode),
+    /// Native Amazon S3 supports copy if not exists through a multipart upload
+    /// where the upload copies an existing object and is completed only if the
+    /// new object does not already exist.
+    ///
+    /// WARNING: When using this mode, `copy_if_not_exists` does not copy tags
+    /// or attributes from the source object.
+    ///
+    /// WARNING: When using this mode, `copy_if_not_exists` makes only a best
+    /// effort attempt to clean up the multipart upload if the copy operation
+    /// fails. Consider using a lifecycle rule to automatically clean up
+    /// abandoned multipart uploads. See [the module
+    /// docs](super#multipart-uploads) for details.
+    ///
+    /// Encoded as `multipart` ignoring whitespace.
+    Multipart,
     /// The name of a DynamoDB table to use for coordination
     ///
     /// Encoded as either `dynamo:<TABLE_NAME>` or `dynamo:<TABLE_NAME>:<TIMEOUT_MILLIS>`
@@ -64,6 +79,7 @@ impl std::fmt::Display for S3CopyIfNotExists {
             Self::HeaderWithStatus(k, v, code) => {
                 write!(f, "header-with-status: {k}: {v}: {}", code.as_u16())
             }
+            Self::Multipart => f.write_str("multipart"),
             Self::Dynamo(lock) => write!(f, "dynamo: {}", lock.table_name()),
         }
     }
@@ -71,6 +87,10 @@ impl std::fmt::Display for S3CopyIfNotExists {
 
 impl S3CopyIfNotExists {
     fn from_str(s: &str) -> Option<Self> {
+        if s.trim() == "multipart" {
+            return Some(Self::Multipart);
+        };
+
         let (variant, value) = s.split_once(':')?;
         match variant.trim() {
             "header" => {
@@ -118,6 +138,17 @@ pub enum S3ConditionalPut {
     /// [HTTP precondition]: https://datatracker.ietf.org/doc/html/rfc9110#name-preconditions
     ETagMatch,
 
+    /// Like `ETagMatch`, but with support for `PutMode::Create` and not
+    /// `PutMode::Option`.
+    ///
+    /// This is the limited form of conditional put supported by Amazon S3
+    /// as of August 2024 ([announcement]).
+    ///
+    /// Encoded as `etag-put-if-not-exists` ignoring whitespace.
+    ///
+    /// [announcement]: https://aws.amazon.com/about-aws/whats-new/2024/08/amazon-s3-conditional-writes/
+    ETagPutIfNotExists,
+
     /// The name of a DynamoDB table to use for coordination
     ///
     /// Encoded as either `dynamo:<TABLE_NAME>` or `dynamo:<TABLE_NAME>:<TIMEOUT_MILLIS>`
@@ -133,6 +164,7 @@ impl std::fmt::Display for S3ConditionalPut {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::ETagMatch => write!(f, "etag"),
+            Self::ETagPutIfNotExists => write!(f, "etag-put-if-not-exists"),
             Self::Dynamo(lock) => write!(f, "dynamo: {}", lock.table_name()),
         }
     }
@@ -142,6 +174,7 @@ impl S3ConditionalPut {
     fn from_str(s: &str) -> Option<Self> {
         match s.trim() {
             "etag" => Some(Self::ETagMatch),
+            "etag-put-if-not-exists" => Some(Self::ETagPutIfNotExists),
             trimmed => match trimmed.split_once(':')? {
                 ("dynamo", s) => Some(Self::Dynamo(DynamoCommit::from_str(s)?)),
                 _ => None,
diff --git a/object_store/src/client/s3.rs b/object_store/src/client/s3.rs
index a9c47260e..dba752cb1 100644
--- a/object_store/src/client/s3.rs
+++ b/object_store/src/client/s3.rs
@@ -92,6 +92,14 @@ pub(crate) struct InitiateMultipartUploadResult {
     pub upload_id: String,
 }
 
+#[cfg(feature = "aws")]
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub(crate) struct CopyPartResult {
+    #[serde(rename = "ETag")]
+    pub e_tag: String,
+}
+
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "PascalCase")]
 pub(crate) struct CompleteMultipartUpload {
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index a259f6bd9..ccc9c341f 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -79,6 +79,9 @@ enum Error {
     #[snafu(display("Error getting put response body: {}", source))]
     PutResponseBody { source: reqwest::Error },
 
+    #[snafu(display("Got invalid put request: {}", source))]
+    InvalidPutRequest { source: quick_xml::se::SeError },
+
     #[snafu(display("Got invalid put response: {}", source))]
     InvalidPutResponse { source: quick_xml::de::DeError },
 
@@ -495,7 +498,7 @@ impl GoogleCloudStorageClient {
         let credential = self.get_credential().await?;
 
         let data = quick_xml::se::to_string(&upload_info)
-            .context(InvalidPutResponseSnafu)?
+            .context(InvalidPutRequestSnafu)?
             // We cannot disable the escaping that transforms "/" to "&quote;" :(
             // https://github.com/tafia/quick-xml/issues/362
             // https://github.com/tafia/quick-xml/issues/350
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
index 89b21bc61..301778783 100644
--- a/object_store/src/integration.rs
+++ b/object_store/src/integration.rs
@@ -651,6 +651,12 @@ pub async fn put_opts(storage: &dyn ObjectStore, supports_update: bool) {
     assert_eq!(b.as_ref(), b"a");
 
     if !supports_update {
+        let err = storage
+            .put_opts(&path, "c".into(), PutMode::Update(v1.clone().into()).into())
+            .await
+            .unwrap_err();
+        assert!(matches!(err, Error::NotImplemented { .. }), "{err}");
+
         return;
     }
 
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 32bc13b62..4064baba0 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -63,7 +63,7 @@ serde_json = { version = "1.0", default-features = false, features = ["std"], op
 seq-macro = { version = "0.3", default-features = false }
 futures = { version = "0.3", default-features = false, features = ["std"], optional = true }
 tokio = { version = "1.0", optional = true, default-features = false, features = ["macros", "rt", "io-util"] }
-hashbrown = { version = "0.14", default-features = false }
+hashbrown = { version = "0.15", default-features = false }
 twox-hash = { version = "1.6", default-features = false }
 paste = { version = "1.0" }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
@@ -81,7 +81,7 @@ lz4_flex = { version = "0.11", default-features = false, features = ["std", "fra
 zstd = { version = "0.13", default-features = false }
 serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
-tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
+tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
 object_store = { version = "0.11.0", default-features = false, features = ["azure"] }
 
diff --git a/parquet/src/arrow/async_reader/store.rs b/parquet/src/arrow/async_reader/store.rs
index e6b47856e..fd0397b5e 100644
--- a/parquet/src/arrow/async_reader/store.rs
+++ b/parquet/src/arrow/async_reader/store.rs
@@ -15,17 +15,15 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::ops::Range;
-use std::sync::Arc;
+use std::{ops::Range, sync::Arc};
 
 use bytes::Bytes;
-use futures::future::BoxFuture;
-use futures::{FutureExt, TryFutureExt};
-
-use object_store::{ObjectMeta, ObjectStore};
+use futures::{future::BoxFuture, FutureExt, TryFutureExt};
+use object_store::{path::Path, ObjectMeta, ObjectStore};
+use tokio::runtime::Handle;
 
 use crate::arrow::async_reader::AsyncFileReader;
-use crate::errors::Result;
+use crate::errors::{ParquetError, Result};
 use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
 
 /// Reads Parquet files in object storage using [`ObjectStore`].
@@ -59,6 +57,7 @@ pub struct ParquetObjectReader {
     metadata_size_hint: Option<usize>,
     preload_column_index: bool,
     preload_offset_index: bool,
+    runtime: Option<Handle>,
 }
 
 impl ParquetObjectReader {
@@ -72,6 +71,7 @@ impl ParquetObjectReader {
             metadata_size_hint: None,
             preload_column_index: false,
             preload_offset_index: false,
+            runtime: None,
         }
     }
 
@@ -99,29 +99,70 @@ impl ParquetObjectReader {
             ..self
         }
     }
+
+    /// Perform IO on the provided tokio runtime
+    ///
+    /// Tokio is a cooperative scheduler, and relies on tasks yielding in a timely manner
+    /// to service IO. Therefore, running IO and CPU-bound tasks, such as parquet decoding,
+    /// on the same tokio runtime can lead to degraded throughput, dropped connections and
+    /// other issues. For more information see [here].
+    ///
+    /// [here]: https://www.influxdata.com/blog/using-rustlangs-async-tokio-runtime-for-cpu-bound-tasks/
+    pub fn with_runtime(self, handle: Handle) -> Self {
+        Self {
+            runtime: Some(handle),
+            ..self
+        }
+    }
+
+    fn spawn<F, O, E>(&self, f: F) -> BoxFuture<'_, Result<O>>
+    where
+        F: for<'a> FnOnce(&'a Arc<dyn ObjectStore>, &'a Path) -> BoxFuture<'a, Result<O, E>>
+            + Send
+            + 'static,
+        O: Send + 'static,
+        E: Into<ParquetError> + Send + 'static,
+    {
+        match &self.runtime {
+            Some(handle) => {
+                let path = self.meta.location.clone();
+                let store = Arc::clone(&self.store);
+                handle
+                    .spawn(async move { f(&store, &path).await })
+                    .map_ok_or_else(
+                        |e| match e.try_into_panic() {
+                            Err(e) => Err(ParquetError::External(Box::new(e))),
+                            Ok(p) => std::panic::resume_unwind(p),
+                        },
+                        |res| res.map_err(|e| e.into()),
+                    )
+                    .boxed()
+            }
+            None => f(&self.store, &self.meta.location)
+                .map_err(|e| e.into())
+                .boxed(),
+        }
+    }
 }
 
 impl AsyncFileReader for ParquetObjectReader {
     fn get_bytes(&mut self, range: Range<usize>) -> BoxFuture<'_, Result<Bytes>> {
-        self.store
-            .get_range(&self.meta.location, range)
-            .map_err(|e| e.into())
-            .boxed()
+        self.spawn(|store, path| store.get_range(path, range))
     }
 
     fn get_byte_ranges(&mut self, ranges: Vec<Range<usize>>) -> BoxFuture<'_, Result<Vec<Bytes>>>
     where
         Self: Send,
     {
-        async move {
-            self.store
-                .get_ranges(&self.meta.location, &ranges)
-                .await
-                .map_err(|e| e.into())
-        }
-        .boxed()
+        self.spawn(|store, path| async move { store.get_ranges(path, &ranges).await }.boxed())
     }
 
+    // This method doesn't directly call `self.spawn` because all of the IO that is done down the
+    // line due to this method call is done through `self.get_bytes` and/or `self.get_byte_ranges`.
+    // When `self` is passed into `ParquetMetaDataReader::load_and_finish`, it treats it as
+    // an `impl MetadataFetch` and calls those methods to get data from it. Due to `Self`'s impl of
+    // `AsyncFileReader`, the calls to `MetadataFetch::fetch` are just delegated to
+    // `Self::get_bytes`.
     fn get_metadata(&mut self) -> BoxFuture<'_, Result<Arc<ParquetMetaData>>> {
         Box::pin(async move {
             let file_size = self.meta.size;
@@ -138,30 +179,40 @@ impl AsyncFileReader for ParquetObjectReader {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    };
 
     use futures::TryStreamExt;
 
     use arrow::util::test_util::parquet_test_data;
+    use futures::FutureExt;
     use object_store::local::LocalFileSystem;
     use object_store::path::Path;
-    use object_store::ObjectStore;
+    use object_store::{ObjectMeta, ObjectStore};
 
-    use crate::arrow::async_reader::ParquetObjectReader;
+    use crate::arrow::async_reader::{AsyncFileReader, ParquetObjectReader};
     use crate::arrow::ParquetRecordBatchStreamBuilder;
+    use crate::errors::ParquetError;
 
-    #[tokio::test]
-    async fn test_simple() {
+    async fn get_meta_store() -> (ObjectMeta, Arc<dyn ObjectStore>) {
         let res = parquet_test_data();
         let store = LocalFileSystem::new_with_prefix(res).unwrap();
 
-        let mut meta = store
+        let meta = store
             .head(&Path::from("alltypes_plain.parquet"))
             .await
             .unwrap();
 
-        let store = Arc::new(store) as Arc<dyn ObjectStore>;
-        let object_reader = ParquetObjectReader::new(Arc::clone(&store), meta.clone());
+        (meta, Arc::new(store) as Arc<dyn ObjectStore>)
+    }
+
+    #[tokio::test]
+    async fn test_simple() {
+        let (meta, store) = get_meta_store().await;
+        let object_reader = ParquetObjectReader::new(store, meta);
+
         let builder = ParquetRecordBatchStreamBuilder::new(object_reader)
             .await
             .unwrap();
@@ -169,7 +220,11 @@ mod tests {
 
         assert_eq!(batches.len(), 1);
         assert_eq!(batches[0].num_rows(), 8);
+    }
 
+    #[tokio::test]
+    async fn test_not_found() {
+        let (mut meta, store) = get_meta_store().await;
         meta.location = Path::from("I don't exist.parquet");
 
         let object_reader = ParquetObjectReader::new(store, meta);
@@ -180,10 +235,86 @@ mod tests {
                 let err = e.to_string();
                 assert!(
                     err.contains("not found: No such file or directory (os error 2)"),
-                    "{}",
-                    err
+                    "{err}",
                 );
             }
         }
     }
+
+    #[tokio::test]
+    async fn test_runtime_is_used() {
+        let num_actions = Arc::new(AtomicUsize::new(0));
+
+        let (a1, a2) = (num_actions.clone(), num_actions.clone());
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .on_thread_park(move || {
+                a1.fetch_add(1, Ordering::Relaxed);
+            })
+            .on_thread_unpark(move || {
+                a2.fetch_add(1, Ordering::Relaxed);
+            })
+            .build()
+            .unwrap();
+
+        let (meta, store) = get_meta_store().await;
+
+        let initial_actions = num_actions.load(Ordering::Relaxed);
+
+        let reader = ParquetObjectReader::new(store, meta).with_runtime(rt.handle().clone());
+
+        let builder = ParquetRecordBatchStreamBuilder::new(reader).await.unwrap();
+        let batches: Vec<_> = builder.build().unwrap().try_collect().await.unwrap();
+
+        // Just copied these assert_eqs from the `test_simple` above
+        assert_eq!(batches.len(), 1);
+        assert_eq!(batches[0].num_rows(), 8);
+
+        assert!(num_actions.load(Ordering::Relaxed) - initial_actions > 0);
+
+        // Runtimes have to be dropped in blocking contexts, so we need to move this one to a new
+        // blocking thread to drop it.
+        tokio::runtime::Handle::current().spawn_blocking(move || drop(rt));
+    }
+
+    /// Unit test that `ParquetObjectReader::spawn`spawns on the provided runtime
+    #[tokio::test]
+    async fn test_runtime_thread_id_different() {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .build()
+            .unwrap();
+
+        let (meta, store) = get_meta_store().await;
+
+        let reader = ParquetObjectReader::new(store, meta).with_runtime(rt.handle().clone());
+
+        let current_id = std::thread::current().id();
+
+        let other_id = reader
+            .spawn(|_, _| async move { Ok::<_, ParquetError>(std::thread::current().id()) }.boxed())
+            .await
+            .unwrap();
+
+        assert_ne!(current_id, other_id);
+
+        tokio::runtime::Handle::current().spawn_blocking(move || drop(rt));
+    }
+
+    #[tokio::test]
+    async fn io_fails_on_shutdown_runtime() {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
+            .build()
+            .unwrap();
+
+        let (meta, store) = get_meta_store().await;
+
+        let mut reader = ParquetObjectReader::new(store, meta).with_runtime(rt.handle().clone());
+
+        rt.shutdown_background();
+
+        let err = reader.get_bytes(0..1).await.unwrap_err().to_string();
+
+        assert!(err.to_string().contains("was cancelled"));
+    }
 }
diff --git a/parquet/src/errors.rs b/parquet/src/errors.rs
index bb4d2543c..6adbffa2a 100644
--- a/parquet/src/errors.rs
+++ b/parquet/src/errors.rs
@@ -106,7 +106,6 @@ impl From<str::Utf8Error> for ParquetError {
         ParquetError::External(Box::new(e))
     }
 }
-
 #[cfg(feature = "arrow")]
 impl From<ArrowError> for ParquetError {
     fn from(e: ArrowError) -> ParquetError {
diff --git a/parquet/src/record/reader.rs b/parquet/src/record/reader.rs
index 1f9128a8b..fd6ca7cdd 100644
--- a/parquet/src/record/reader.rs
+++ b/parquet/src/record/reader.rs
@@ -138,7 +138,17 @@ impl TreeBuilder {
                 .column_descr_ptr();
             let col_reader = row_group_reader.get_column_reader(orig_index)?;
             let column = TripletIter::new(col_descr, col_reader, self.batch_size);
-            Reader::PrimitiveReader(field, Box::new(column))
+            let reader = Reader::PrimitiveReader(field.clone(), Box::new(column));
+            if repetition == Repetition::REPEATED {
+                Reader::RepeatedReader(
+                    field,
+                    curr_def_level - 1,
+                    curr_rep_level - 1,
+                    Box::new(reader),
+                )
+            } else {
+                reader
+            }
         } else {
             match field.get_basic_info().converted_type() {
                 // List types
@@ -1688,6 +1698,131 @@ mod tests {
         assert_eq!(rows, expected_rows);
     }
 
+    #[test]
+    fn test_tree_reader_handle_primitive_repeated_fields_with_no_annotation() {
+        // In this test the REPEATED fields are primitives
+        let rows = test_file_reader_rows("repeated_primitive_no_list.parquet", None).unwrap();
+        let expected_rows = vec![
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list([0, 1, 2, 3].map(Field::Int).to_vec()))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["foo", "zero", "one", "two"]
+                            .map(|s| Field::Str(s.to_string()))
+                            .to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list([0, 1, 2, 3].map(Field::Int).to_vec()))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["foo", "zero", "one", "two"]
+                                    .map(|s| Field::Str(s.to_string()))
+                                    .to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list(vec![]))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["three"].map(|s| Field::Str(s.to_string())).to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list(vec![]))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["three"].map(|s| Field::Str(s.to_string())).to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list(vec![Field::Int(4)]))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["four"].map(|s| Field::Str(s.to_string())).to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list(vec![Field::Int(4)]))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["four"].map(|s| Field::Str(s.to_string())).to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+            row![
+                (
+                    "Int32_list".to_string(),
+                    Field::ListInternal(make_list([5, 6, 7, 8].map(Field::Int).to_vec()))
+                ),
+                (
+                    "String_list".to_string(),
+                    Field::ListInternal(make_list(
+                        ["five", "six", "seven", "eight"]
+                            .map(|s| Field::Str(s.to_string()))
+                            .to_vec()
+                    ))
+                ),
+                (
+                    "group_of_lists".to_string(),
+                    group![
+                        (
+                            "Int32_list_in_group".to_string(),
+                            Field::ListInternal(make_list([5, 6, 7, 8].map(Field::Int).to_vec()))
+                        ),
+                        (
+                            "String_list_in_group".to_string(),
+                            Field::ListInternal(make_list(
+                                ["five", "six", "seven", "eight"]
+                                    .map(|s| Field::Str(s.to_string()))
+                                    .to_vec()
+                            ))
+                        )
+                    ]
+                )
+            ],
+        ];
+        assert_eq!(rows, expected_rows);
+    }
+
     fn test_file_reader_rows(file_name: &str, schema: Option<Type>) -> Result<Vec<Row>> {
         let file = get_test_file(file_name);
         let file_reader: Box<dyn FileReader> = Box::new(SerializedFileReader::new(file)?);
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index dc67abe05..b7ba95eb5 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -1134,7 +1134,7 @@ fn from_thrift_helper(elements: &[SchemaElement], index: usize) -> Result<(usize
     // There is only one message type node in the schema tree.
     let is_root_node = index == 0;
 
-    if index > elements.len() {
+    if index >= elements.len() {
         return Err(general_err!(
             "Index out of bound, index = {}, len = {}",
             index,


Test patch:
diff --git a/parquet-testing b/parquet-testing
index 50af3d8ce..550368ca7 160000
--- a/parquet-testing
+++ b/parquet-testing
@@ -1,1 +1,1 @@
-Subproject commit 50af3d8ce206990d81014b1862e5ce7380dc3e08
+Subproject commit 550368ca77b97231efead39251a96bd6f8f08c6e