Repository: apache/arrow-rs
Pull request #5604 (2024-04-08 13:10:42)

Update prost-build requirement from =0.12.3 to =0.12.4
Updates the requirements on [prost-build](https://github.com/tokio-rs/prost) to permit the latest version.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/tokio-rs/prost/commit/38a00d89527728b6d50736d2ede49b5963abc2fd"><code>38a00d8</code></a> chore: Release version 0.12.4 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1024">#1024</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/b3de938c21e9875f17e3a4c116203f9c2f8e8640"><code>b3de938</code></a> docs: Document feature flags (<a href="https://redirect.github.com/tokio-rs/prost/issues/1003">#1003</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/12260d1f8409ea4ec86d0909ab2d49174e1a0468"><code>12260d1</code></a> build(deps): Allow multimap 0.10 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1013">#1013</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/2b6140cf6fe899e9817816bae97d19fe8e0eab60"><code>2b6140c</code></a> build(deps): Allow heck 0.5 (<a href="https://redirect.github.com/tokio-rs/prost/issues/1012">#1012</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/50bab4ff9f5bae1cde6b54d6de02107d30f11453"><code>50bab4f</code></a> chore: Split prost-types lib.rs into separate modules. (<a href="https://redirect.github.com/tokio-rs/prost/issues/1007">#1007</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/0bd94826f04b7ead7dc6a0ad2ac749374cd2dfd3"><code>0bd9482</code></a> Minor clippy lint fixes. (<a href="https://redirect.github.com/tokio-rs/prost/issues/1006">#1006</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/82d777491aba5118d0f9562d31bf2eae0cff4ab6"><code>82d7774</code></a> Allow itertools 0.12 (<a href="https://redirect.github.com/tokio-rs/prost/issues/948">#948</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/2a4aeafa31dddbfe7d8dd7915c561f95ef45ec93"><code>2a4aeaf</code></a> feat: add derive feature (<a href="https://redirect.github.com/tokio-rs/prost/issues/992">#992</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/e3d708c0e031d71b87bb5ba5c0c3aad3393275fd"><code>e3d708c</code></a> Improve protoc not found error message (<a href="https://redirect.github.com/tokio-rs/prost/issues/937">#937</a>)</li>
<li><a href="https://github.com/tokio-rs/prost/commit/145fd47204df53944e2d1ba964d326f7081253cb"><code>145fd47</code></a> improve encode_varint performance by bounding its loop (<a href="https://redirect.github.com/tokio-rs/prost/issues/940">#940</a>)</li>
<li>Additional commits viewable in <a href="https://github.com/tokio-rs/prost/compare/v0.12.3...v0.12.4">compare view</a></li>
</ul>
</details>
<br />


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/arrow-array/src/array/primitive_array.rs b/arrow-array/src/array/primitive_array.rs
index ddae770d3..924cab1ac 100644
--- a/arrow-array/src/array/primitive_array.rs
+++ b/arrow-array/src/array/primitive_array.rs
@@ -1097,7 +1097,7 @@ impl<T: ArrowPrimitiveType> std::fmt::Debug for PrimitiveArray<T> {
         write!(f, "PrimitiveArray<{data_type:?}>\n[\n")?;
         print_long_array(self, f, |array, index, f| match data_type {
             DataType::Date32 | DataType::Date64 => {
-                let v = self.value(index).to_isize().unwrap() as i64;
+                let v = self.value(index).to_i64().unwrap();
                 match as_date::<T>(v) {
                     Some(date) => write!(f, "{date:?}"),
                     None => {
@@ -1109,7 +1109,7 @@ impl<T: ArrowPrimitiveType> std::fmt::Debug for PrimitiveArray<T> {
                 }
             }
             DataType::Time32(_) | DataType::Time64(_) => {
-                let v = self.value(index).to_isize().unwrap() as i64;
+                let v = self.value(index).to_i64().unwrap();
                 match as_time::<T>(v) {
                     Some(time) => write!(f, "{time:?}"),
                     None => {
@@ -1121,7 +1121,7 @@ impl<T: ArrowPrimitiveType> std::fmt::Debug for PrimitiveArray<T> {
                 }
             }
             DataType::Timestamp(_, tz_string_opt) => {
-                let v = self.value(index).to_isize().unwrap() as i64;
+                let v = self.value(index).to_i64().unwrap();
                 match tz_string_opt {
                     // for Timestamp with TimeZone
                     Some(tz_string) => {
diff --git a/arrow-array/src/array/union_array.rs b/arrow-array/src/array/union_array.rs
index e3e637247..22d4cf90a 100644
--- a/arrow-array/src/array/union_array.rs
+++ b/arrow-array/src/array/union_array.rs
@@ -23,6 +23,7 @@ use arrow_schema::{ArrowError, DataType, Field, UnionFields, UnionMode};
 /// Contains the `UnionArray` type.
 ///
 use std::any::Any;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 /// An array of [values of varying types](https://arrow.apache.org/docs/format/Columnar.html#union-layout)
@@ -319,6 +320,70 @@ impl UnionArray {
             fields,
         }
     }
+
+    /// Deconstruct this array into its constituent parts
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use arrow_array::array::UnionArray;
+    /// # use arrow_array::types::Int32Type;
+    /// # use arrow_array::builder::UnionBuilder;
+    /// # use arrow_buffer::ScalarBuffer;
+    /// # fn main() -> Result<(), arrow_schema::ArrowError> {
+    /// let mut builder = UnionBuilder::new_dense();
+    /// builder.append::<Int32Type>("a", 1).unwrap();
+    /// let union_array = builder.build()?;
+    ///
+    /// // Deconstruct into parts
+    /// let (type_ids, offsets, field_type_ids, fields) = union_array.into_parts();
+    ///
+    /// // Reconstruct from parts
+    /// let union_array = UnionArray::try_new(
+    ///     &field_type_ids,
+    ///     type_ids.into_inner(),
+    ///     offsets.map(ScalarBuffer::into_inner),
+    ///     fields,
+    /// );
+    /// # Ok(())
+    /// # }
+    /// ```
+    #[allow(clippy::type_complexity)]
+    pub fn into_parts(
+        self,
+    ) -> (
+        ScalarBuffer<i8>,
+        Option<ScalarBuffer<i32>>,
+        Vec<i8>,
+        Vec<(Field, ArrayRef)>,
+    ) {
+        let Self {
+            data_type,
+            type_ids,
+            offsets,
+            fields,
+        } = self;
+        match data_type {
+            DataType::Union(union_fields, _) => {
+                let union_fields = union_fields.iter().collect::<HashMap<_, _>>();
+                let (field_type_ids, fields) = fields
+                    .into_iter()
+                    .enumerate()
+                    .flat_map(|(type_id, array_ref)| {
+                        array_ref.map(|array_ref| {
+                            let type_id = type_id as i8;
+                            (
+                                type_id,
+                                ((*Arc::clone(union_fields[&type_id])).clone(), array_ref),
+                            )
+                        })
+                    })
+                    .unzip();
+                (type_ids, offsets, field_type_ids, fields)
+            }
+            _ => unreachable!(),
+        }
+    }
 }
 
 impl From<ArrayData> for UnionArray {
@@ -505,6 +570,7 @@ impl std::fmt::Debug for UnionArray {
 mod tests {
     use super::*;
 
+    use crate::array::Int8Type;
     use crate::builder::UnionBuilder;
     use crate::cast::AsArray;
     use crate::types::{Float32Type, Float64Type, Int32Type, Int64Type};
@@ -1201,4 +1267,104 @@ mod tests {
         assert_eq!(v.len(), 1);
         assert_eq!(v.as_string::<i32>().value(0), "baz");
     }
+
+    #[test]
+    fn into_parts() {
+        let mut builder = UnionBuilder::new_dense();
+        builder.append::<Int32Type>("a", 1).unwrap();
+        builder.append::<Int8Type>("b", 2).unwrap();
+        builder.append::<Int32Type>("a", 3).unwrap();
+        let dense_union = builder.build().unwrap();
+
+        let field = [
+            Field::new("a", DataType::Int32, false),
+            Field::new("b", DataType::Int8, false),
+        ];
+        let (type_ids, offsets, field_type_ids, fields) = dense_union.into_parts();
+        assert_eq!(field_type_ids, [0, 1]);
+        assert_eq!(
+            field.to_vec(),
+            fields
+                .iter()
+                .cloned()
+                .map(|(field, _)| field)
+                .collect::<Vec<_>>()
+        );
+        assert_eq!(type_ids, [0, 1, 0]);
+        assert!(offsets.is_some());
+        assert_eq!(offsets.as_ref().unwrap(), &[0, 0, 1]);
+
+        let result = UnionArray::try_new(
+            &field_type_ids,
+            type_ids.into_inner(),
+            offsets.map(ScalarBuffer::into_inner),
+            fields,
+        );
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap().len(), 3);
+
+        let mut builder = UnionBuilder::new_sparse();
+        builder.append::<Int32Type>("a", 1).unwrap();
+        builder.append::<Int8Type>("b", 2).unwrap();
+        builder.append::<Int32Type>("a", 3).unwrap();
+        let sparse_union = builder.build().unwrap();
+
+        let (type_ids, offsets, field_type_ids, fields) = sparse_union.into_parts();
+        assert_eq!(type_ids, [0, 1, 0]);
+        assert!(offsets.is_none());
+
+        let result = UnionArray::try_new(
+            &field_type_ids,
+            type_ids.into_inner(),
+            offsets.map(ScalarBuffer::into_inner),
+            fields,
+        );
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap().len(), 3);
+    }
+
+    #[test]
+    fn into_parts_custom_type_ids() {
+        let mut set_field_type_ids: [i8; 3] = [8, 4, 9];
+        let data_type = DataType::Union(
+            UnionFields::new(
+                set_field_type_ids,
+                [
+                    Field::new("strings", DataType::Utf8, false),
+                    Field::new("integers", DataType::Int32, false),
+                    Field::new("floats", DataType::Float64, false),
+                ],
+            ),
+            UnionMode::Dense,
+        );
+        let string_array = StringArray::from(vec!["foo", "bar", "baz"]);
+        let int_array = Int32Array::from(vec![5, 6, 4]);
+        let float_array = Float64Array::from(vec![10.0]);
+        let type_ids = Buffer::from_vec(vec![4_i8, 8, 4, 8, 9, 4, 8]);
+        let value_offsets = Buffer::from_vec(vec![0_i32, 0, 1, 1, 0, 2, 2]);
+        let data = ArrayData::builder(data_type)
+            .len(7)
+            .buffers(vec![type_ids, value_offsets])
+            .child_data(vec![
+                string_array.into_data(),
+                int_array.into_data(),
+                float_array.into_data(),
+            ])
+            .build()
+            .unwrap();
+        let array = UnionArray::from(data);
+
+        let (type_ids, offsets, field_type_ids, fields) = array.into_parts();
+        set_field_type_ids.sort();
+        assert_eq!(field_type_ids, set_field_type_ids);
+        let result = UnionArray::try_new(
+            &field_type_ids,
+            type_ids.into_inner(),
+            offsets.map(ScalarBuffer::into_inner),
+            fields,
+        );
+        assert!(result.is_ok());
+        let array = result.unwrap();
+        assert_eq!(array.len(), 7);
+    }
 }
diff --git a/arrow-array/src/builder/fixed_size_list_builder.rs b/arrow-array/src/builder/fixed_size_list_builder.rs
index 0fe779d5c..3d79a763e 100644
--- a/arrow-array/src/builder/fixed_size_list_builder.rs
+++ b/arrow-array/src/builder/fixed_size_list_builder.rs
@@ -19,7 +19,7 @@ use crate::builder::ArrayBuilder;
 use crate::{ArrayRef, FixedSizeListArray};
 use arrow_buffer::NullBufferBuilder;
 use arrow_data::ArrayData;
-use arrow_schema::{DataType, Field};
+use arrow_schema::{DataType, Field, FieldRef};
 use std::any::Any;
 use std::sync::Arc;
 
@@ -67,6 +67,7 @@ pub struct FixedSizeListBuilder<T: ArrayBuilder> {
     null_buffer_builder: NullBufferBuilder,
     values_builder: T,
     list_len: i32,
+    field: Option<FieldRef>,
 }
 
 impl<T: ArrayBuilder> FixedSizeListBuilder<T> {
@@ -89,6 +90,20 @@ impl<T: ArrayBuilder> FixedSizeListBuilder<T> {
             null_buffer_builder: NullBufferBuilder::new(capacity),
             values_builder,
             list_len: value_length,
+            field: None,
+        }
+    }
+
+    /// Override the field passed to [`ArrayData::builder`]
+    ///
+    /// By default a nullable field is created with the name `item`
+    ///
+    /// Note: [`Self::finish`] and [`Self::finish_cloned`] will panic if the
+    /// field's data type does not match that of `T`
+    pub fn with_field(self, field: impl Into<FieldRef>) -> Self {
+        Self {
+            field: Some(field.into()),
+            ..self
         }
     }
 }
@@ -166,13 +181,40 @@ where
         );
 
         let nulls = self.null_buffer_builder.finish();
-        let array_data = ArrayData::builder(DataType::FixedSizeList(
-            Arc::new(Field::new("item", values_data.data_type().clone(), true)),
-            self.list_len,
-        ))
-        .len(len)
-        .add_child_data(values_data)
-        .nulls(nulls);
+
+        let field = match &self.field {
+            Some(f) => {
+                let size = self.value_length();
+                assert_eq!(
+                    f.data_type(),
+                    values_data.data_type(),
+                    "DataType of field ({}) should be the same as the values_builder DataType ({})",
+                    f.data_type(),
+                    values_data.data_type()
+                );
+
+                if let Some(a) = values_arr.logical_nulls() {
+                    let nulls_valid = f.is_nullable()
+                        || nulls
+                            .as_ref()
+                            .map(|n| n.expand(size as _).contains(&a))
+                            .unwrap_or_default();
+
+                    assert!(
+                        nulls_valid,
+                        "Found unmasked nulls for non-nullable FixedSizeListBuilder field {:?}",
+                        f.name()
+                    );
+                }
+                f.clone()
+            }
+            None => Arc::new(Field::new("item", values_data.data_type().clone(), true)),
+        };
+
+        let array_data = ArrayData::builder(DataType::FixedSizeList(field, self.list_len))
+            .len(len)
+            .add_child_data(values_data)
+            .nulls(nulls);
 
         let array_data = unsafe { array_data.build_unchecked() };
 
@@ -194,13 +236,39 @@ where
         );
 
         let nulls = self.null_buffer_builder.finish_cloned();
-        let array_data = ArrayData::builder(DataType::FixedSizeList(
-            Arc::new(Field::new("item", values_data.data_type().clone(), true)),
-            self.list_len,
-        ))
-        .len(len)
-        .add_child_data(values_data)
-        .nulls(nulls);
+
+        let field = match &self.field {
+            Some(f) => {
+                let size = self.value_length();
+                assert_eq!(
+                    f.data_type(),
+                    values_data.data_type(),
+                    "DataType of field ({}) should be the same as the values_builder DataType ({})",
+                    f.data_type(),
+                    values_data.data_type()
+                );
+                if let Some(a) = values_arr.logical_nulls() {
+                    let nulls_valid = f.is_nullable()
+                        || nulls
+                            .as_ref()
+                            .map(|n| n.expand(size as _).contains(&a))
+                            .unwrap_or_default();
+
+                    assert!(
+                        nulls_valid,
+                        "Found unmasked nulls for non-nullable FixedSizeListBuilder field {:?}",
+                        f.name()
+                    );
+                }
+                f.clone()
+            }
+            None => Arc::new(Field::new("item", values_data.data_type().clone(), true)),
+        };
+
+        let array_data = ArrayData::builder(DataType::FixedSizeList(field, self.list_len))
+            .len(len)
+            .add_child_data(values_data)
+            .nulls(nulls);
 
         let array_data = unsafe { array_data.build_unchecked() };
 
@@ -216,28 +284,54 @@ mod tests {
     use crate::Array;
     use crate::Int32Array;
 
-    #[test]
-    fn test_fixed_size_list_array_builder() {
+    fn make_list_builder(
+        include_null_element: bool,
+        include_null_in_values: bool,
+    ) -> FixedSizeListBuilder<crate::builder::PrimitiveBuilder<crate::types::Int32Type>> {
         let values_builder = Int32Builder::new();
         let mut builder = FixedSizeListBuilder::new(values_builder, 3);
 
-        //  [[0, 1, 2], null, [3, null, 5], [6, 7, null]]
         builder.values().append_value(0);
         builder.values().append_value(1);
         builder.values().append_value(2);
         builder.append(true);
-        builder.values().append_null();
-        builder.values().append_null();
-        builder.values().append_null();
-        builder.append(false);
+
+        builder.values().append_value(2);
         builder.values().append_value(3);
-        builder.values().append_null();
-        builder.values().append_value(5);
-        builder.append(true);
-        builder.values().append_value(6);
-        builder.values().append_value(7);
-        builder.values().append_null();
+        builder.values().append_value(4);
         builder.append(true);
+
+        if include_null_element {
+            builder.values().append_null();
+            builder.values().append_null();
+            builder.values().append_null();
+            builder.append(false);
+        } else {
+            builder.values().append_value(2);
+            builder.values().append_value(3);
+            builder.values().append_value(4);
+            builder.append(true);
+        }
+
+        if include_null_in_values {
+            builder.values().append_value(3);
+            builder.values().append_null();
+            builder.values().append_value(5);
+            builder.append(true);
+        } else {
+            builder.values().append_value(3);
+            builder.values().append_value(4);
+            builder.values().append_value(5);
+            builder.append(true);
+        }
+
+        builder
+    }
+
+    #[test]
+    fn test_fixed_size_list_array_builder() {
+        let mut builder = make_list_builder(true, true);
+
         let list_array = builder.finish();
 
         assert_eq!(DataType::Int32, list_array.value_type());
@@ -248,9 +342,48 @@ mod tests {
     }
 
     #[test]
-    fn test_fixed_size_list_array_builder_finish_cloned() {
+    fn test_fixed_size_list_array_builder_with_field() {
+        let builder = make_list_builder(false, false);
+        let mut builder = builder.with_field(Field::new("list_element", DataType::Int32, false));
+        let list_array = builder.finish();
+
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(4, list_array.len());
+        assert_eq!(0, list_array.null_count());
+        assert_eq!(6, list_array.value_offset(2));
+        assert_eq!(3, list_array.value_length());
+    }
+
+    #[test]
+    fn test_fixed_size_list_array_builder_with_field_and_null() {
+        let builder = make_list_builder(true, false);
+        let mut builder = builder.with_field(Field::new("list_element", DataType::Int32, false));
+        let list_array = builder.finish();
+
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(4, list_array.len());
+        assert_eq!(1, list_array.null_count());
+        assert_eq!(6, list_array.value_offset(2));
+        assert_eq!(3, list_array.value_length());
+    }
+
+    #[test]
+    #[should_panic(expected = "Found unmasked nulls for non-nullable FixedSizeListBuilder field")]
+    fn test_fixed_size_list_array_builder_with_field_null_panic() {
+        let builder = make_list_builder(true, true);
+        let mut builder = builder.with_field(Field::new("list_item", DataType::Int32, false));
+
+        builder.finish();
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "DataType of field (Int64) should be the same as the values_builder DataType (Int32)"
+    )]
+    fn test_fixed_size_list_array_builder_with_field_type_panic() {
         let values_builder = Int32Builder::new();
-        let mut builder = FixedSizeListBuilder::new(values_builder, 3);
+        let builder = FixedSizeListBuilder::new(values_builder, 3);
+        let mut builder = builder.with_field(Field::new("list_item", DataType::Int64, true));
 
         //  [[0, 1, 2], null, [3, null, 5], [6, 7, null]]
         builder.values().append_value(0);
@@ -262,13 +395,68 @@ mod tests {
         builder.values().append_null();
         builder.append(false);
         builder.values().append_value(3);
-        builder.values().append_null();
+        builder.values().append_value(4);
         builder.values().append_value(5);
         builder.append(true);
+
+        builder.finish();
+    }
+
+    #[test]
+    fn test_fixed_size_list_array_builder_cloned_with_field() {
+        let builder = make_list_builder(true, true);
+        let builder = builder.with_field(Field::new("list_element", DataType::Int32, true));
+
+        let list_array = builder.finish_cloned();
+
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(4, list_array.len());
+        assert_eq!(1, list_array.null_count());
+        assert_eq!(6, list_array.value_offset(2));
+        assert_eq!(3, list_array.value_length());
+    }
+
+    #[test]
+    #[should_panic(expected = "Found unmasked nulls for non-nullable FixedSizeListBuilder field")]
+    fn test_fixed_size_list_array_builder_cloned_with_field_null_panic() {
+        let builder = make_list_builder(true, true);
+        let builder = builder.with_field(Field::new("list_item", DataType::Int32, false));
+
+        builder.finish_cloned();
+    }
+
+    #[test]
+    fn test_fixed_size_list_array_builder_cloned_with_field_and_null() {
+        let builder = make_list_builder(true, false);
+        let mut builder = builder.with_field(Field::new("list_element", DataType::Int32, false));
+        let list_array = builder.finish();
+
+        assert_eq!(DataType::Int32, list_array.value_type());
+        assert_eq!(4, list_array.len());
+        assert_eq!(1, list_array.null_count());
+        assert_eq!(6, list_array.value_offset(2));
+        assert_eq!(3, list_array.value_length());
+    }
+
+    #[test]
+    #[should_panic(
+        expected = "DataType of field (Int64) should be the same as the values_builder DataType (Int32)"
+    )]
+    fn test_fixed_size_list_array_builder_cloned_with_field_type_panic() {
+        let builder = make_list_builder(false, false);
+        let builder = builder.with_field(Field::new("list_item", DataType::Int64, true));
+
+        builder.finish_cloned();
+    }
+
+    #[test]
+    fn test_fixed_size_list_array_builder_finish_cloned() {
+        let mut builder = make_list_builder(true, true);
+
         let mut list_array = builder.finish_cloned();
 
         assert_eq!(DataType::Int32, list_array.value_type());
-        assert_eq!(3, list_array.len());
+        assert_eq!(4, list_array.len());
         assert_eq!(1, list_array.null_count());
         assert_eq!(3, list_array.value_length());
 
@@ -283,12 +471,40 @@ mod tests {
         list_array = builder.finish();
 
         assert_eq!(DataType::Int32, list_array.value_type());
-        assert_eq!(5, list_array.len());
+        assert_eq!(6, list_array.len());
         assert_eq!(2, list_array.null_count());
         assert_eq!(6, list_array.value_offset(2));
         assert_eq!(3, list_array.value_length());
     }
 
+    #[test]
+    fn test_fixed_size_list_array_builder_with_field_empty() {
+        let values_builder = Int32Array::builder(0);
+        let mut builder = FixedSizeListBuilder::new(values_builder, 3).with_field(Field::new(
+            "list_item",
+            DataType::Int32,
+            false,
+        ));
+        assert!(builder.is_empty());
+        let arr = builder.finish();
+        assert_eq!(0, arr.len());
+        assert_eq!(0, builder.len());
+    }
+
+    #[test]
+    fn test_fixed_size_list_array_builder_cloned_with_field_empty() {
+        let values_builder = Int32Array::builder(0);
+        let builder = FixedSizeListBuilder::new(values_builder, 3).with_field(Field::new(
+            "list_item",
+            DataType::Int32,
+            false,
+        ));
+        assert!(builder.is_empty());
+        let arr = builder.finish_cloned();
+        assert_eq!(0, arr.len());
+        assert_eq!(0, builder.len());
+    }
+
     #[test]
     fn test_fixed_size_list_array_builder_empty() {
         let values_builder = Int32Array::builder(5);
diff --git a/arrow-array/src/builder/generic_bytes_builder.rs b/arrow-array/src/builder/generic_bytes_builder.rs
index 9939a85f9..516f45be3 100644
--- a/arrow-array/src/builder/generic_bytes_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_builder.rs
@@ -26,6 +26,9 @@ use std::fmt::Write;
 use std::sync::Arc;
 
 /// Builder for [`GenericByteArray`]
+///
+/// For building strings, see docs on [`GenericStringBuilder`].
+/// For building binary, see docs on [`GenericBinaryBuilder`].
 pub struct GenericByteBuilder<T: ByteArrayType> {
     value_builder: UInt8BufferBuilder,
     offsets_builder: BufferBuilder<T::Offset>,
@@ -222,11 +225,12 @@ impl<T: ByteArrayType, V: AsRef<T::Native>> Extend<Option<V>> for GenericByteBui
 /// Array builder for [`GenericStringArray`][crate::GenericStringArray]
 ///
 /// Values can be appended using [`GenericByteBuilder::append_value`], and nulls with
-/// [`GenericByteBuilder::append_null`] as normal.
+/// [`GenericByteBuilder::append_null`].
 ///
-/// Additionally implements [`std::fmt::Write`] with any written data included in the next
+/// 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
 ///
+/// # Example
 /// ```
 /// # use std::fmt::Write;
 /// # use arrow_array::builder::GenericStringBuilder;
@@ -257,6 +261,26 @@ impl<O: OffsetSizeTrait> Write for GenericStringBuilder<O> {
 }
 
 ///  Array builder for [`GenericBinaryArray`][crate::GenericBinaryArray]
+///
+/// Values can be appended using [`GenericByteBuilder::append_value`], and nulls with
+/// [`GenericByteBuilder::append_null`].
+///
+/// # Example
+/// ```
+/// # use arrow_array::builder::GenericBinaryBuilder;
+/// let mut builder = GenericBinaryBuilder::<i32>::new();
+///
+/// // Write data
+/// builder.append_value("foo");
+///
+/// // Write second value
+/// builder.append_value(&[0,1,2]);
+///
+/// let array = builder.finish();
+/// // binary values
+/// assert_eq!(array.value(0), b"foo");
+/// assert_eq!(array.value(1), b"\x00\x01\x02");
+/// ```
 pub type GenericBinaryBuilder<O> = GenericByteBuilder<GenericBinaryType<O>>;
 
 #[cfg(test)]
diff --git a/arrow-array/src/builder/mod.rs b/arrow-array/src/builder/mod.rs
index e4ab7ae4b..dd1a5c3ae 100644
--- a/arrow-array/src/builder/mod.rs
+++ b/arrow-array/src/builder/mod.rs
@@ -305,13 +305,21 @@ pub type ListBuilder<T> = GenericListBuilder<i32, T>;
 pub type LargeListBuilder<T> = GenericListBuilder<i64, T>;
 
 /// Builder for [`BinaryArray`](crate::array::BinaryArray)
+///
+/// See examples on [`GenericBinaryBuilder`]
 pub type BinaryBuilder = GenericBinaryBuilder<i32>;
 
 /// Builder for [`LargeBinaryArray`](crate::array::LargeBinaryArray)
+///
+/// See examples on [`GenericBinaryBuilder`]
 pub type LargeBinaryBuilder = GenericBinaryBuilder<i64>;
 
 /// Builder for [`StringArray`](crate::array::StringArray)
+///
+/// See examples on [`GenericStringBuilder`]
 pub type StringBuilder = GenericStringBuilder<i32>;
 
 /// Builder for [`LargeStringArray`](crate::array::LargeStringArray)
+///
+/// See examples on [`GenericStringBuilder`]
 pub type LargeStringBuilder = GenericStringBuilder<i64>;
diff --git a/arrow-array/src/cast.rs b/arrow-array/src/cast.rs
index 2e21f3e7e..7b4b1d6ec 100644
--- a/arrow-array/src/cast.rs
+++ b/arrow-array/src/cast.rs
@@ -779,6 +779,34 @@ 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 [`BinaryViewArray`] returning `None` if not possible
+    fn as_binary_view_opt(&self) -> Option<&BinaryViewArray> {
+        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 [`GenericByteViewArray`] returning `None` if not possible
+    fn as_byte_view_opt<T: ByteViewType>(&self) -> Option<&GenericByteViewArray<T>>;
+
     /// Downcast this to a [`StructArray`] returning `None` if not possible
     fn as_struct_opt(&self) -> Option<&StructArray>;
 
@@ -852,6 +880,10 @@ impl AsArray for dyn Array + '_ {
         self.as_any().downcast_ref()
     }
 
+    fn as_byte_view_opt<T: ByteViewType>(&self) -> Option<&GenericByteViewArray<T>> {
+        self.as_any().downcast_ref()
+    }
+
     fn as_struct_opt(&self) -> Option<&StructArray> {
         self.as_any().downcast_ref()
     }
@@ -899,6 +931,10 @@ impl AsArray for ArrayRef {
         self.as_ref().as_bytes_opt()
     }
 
+    fn as_byte_view_opt<T: ByteViewType>(&self) -> Option<&GenericByteViewArray<T>> {
+        self.as_ref().as_byte_view_opt()
+    }
+
     fn as_struct_opt(&self) -> Option<&StructArray> {
         self.as_ref().as_struct_opt()
     }
diff --git a/arrow-buffer/src/native.rs b/arrow-buffer/src/native.rs
index 5184d60ac..680974351 100644
--- a/arrow-buffer/src/native.rs
+++ b/arrow-buffer/src/native.rs
@@ -75,6 +75,12 @@ pub trait ArrowNativeType:
     /// in truncation/overflow
     fn to_isize(self) -> Option<isize>;
 
+    /// Convert native type to i64.
+    ///
+    /// Returns `None` if [`Self`] is not an integer or conversion would result
+    /// in truncation/overflow
+    fn to_i64(self) -> Option<i64>;
+
     /// Convert native type from i32.
     ///
     /// Returns `None` if [`Self`] is not `i32`
@@ -119,6 +125,11 @@ macro_rules! native_integer {
                 self.try_into().ok()
             }
 
+            #[inline]
+            fn to_i64(self) -> Option<i64> {
+                self.try_into().ok()
+            }
+
             #[inline]
             fn as_usize(self) -> usize {
                 self as _
@@ -170,6 +181,11 @@ macro_rules! native_float {
                 None
             }
 
+            #[inline]
+            fn to_i64(self) -> Option<i64> {
+                None
+            }
+
             #[inline]
             fn as_usize($s) -> usize {
                 $as_usize
@@ -212,6 +228,10 @@ impl ArrowNativeType for i256 {
     fn to_isize(self) -> Option<isize> {
         self.to_i128()?.try_into().ok()
     }
+
+    fn to_i64(self) -> Option<i64> {
+        self.to_i128()?.try_into().ok()
+    }
 }
 
 /// Allows conversion from supported Arrow types to a byte slice.
diff --git a/arrow-flight/gen/Cargo.toml b/arrow-flight/gen/Cargo.toml
index 74e636426..fcf81a845 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.79", default-features = false }
-prost-build = { version = "=0.12.3", default-features = false }
+prost-build = { version = "=0.12.4", default-features = false }
 tonic-build = { version = "=0.11.0", 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 67265e781..bc314de9d 100644
--- a/arrow-flight/src/arrow.flight.protocol.rs
+++ b/arrow-flight/src/arrow.flight.protocol.rs
@@ -1,5 +1,6 @@
 // This file was automatically generated through the build.rs script, and should not be edited.
 
+// 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)]
diff --git a/arrow-flight/src/sql/arrow.flight.protocol.sql.rs b/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
index 01ea9b61a..c1f0fac0f 100644
--- a/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
+++ b/arrow-flight/src/sql/arrow.flight.protocol.sql.rs
@@ -1,5 +1,6 @@
 // This file was automatically generated through the build.rs script, and should not be edited.
 
+// This file is @generated by prost-build.
 ///
 /// Represents a metadata request. Used in the command member of FlightDescriptor
 /// for the following RPC calls:
diff --git a/arrow-select/src/take.rs b/arrow-select/src/take.rs
index d9a639da8..dc9e13040 100644
--- a/arrow-select/src/take.rs
+++ b/arrow-select/src/take.rs
@@ -143,6 +143,9 @@ fn take_impl<IndexType: ArrowPrimitiveType>(
         DataType::LargeUtf8 => {
             Ok(Arc::new(take_bytes(values.as_string::<i64>(), indices)?))
         }
+        DataType::Utf8View => {
+            Ok(Arc::new(take_byte_view(values.as_string_view(), indices)?))
+        }
         DataType::List(_) => {
             Ok(Arc::new(take_list::<_, Int32Type>(values.as_list(), indices)?))
         }
@@ -204,6 +207,9 @@ fn take_impl<IndexType: ArrowPrimitiveType>(
         DataType::LargeBinary => {
             Ok(Arc::new(take_bytes(values.as_binary::<i64>(), indices)?))
         }
+        DataType::BinaryView => {
+            Ok(Arc::new(take_byte_view(values.as_binary_view(), indices)?))
+        }
         DataType::FixedSizeBinary(size) => {
             let values = values
                 .as_any()
@@ -437,6 +443,20 @@ fn take_bytes<T: ByteArrayType, IndexType: ArrowPrimitiveType>(
     Ok(GenericByteArray::from(array_data))
 }
 
+/// `take` implementation for byte view arrays
+fn take_byte_view<T: ByteViewType, IndexType: ArrowPrimitiveType>(
+    array: &GenericByteViewArray<T>,
+    indices: &PrimitiveArray<IndexType>,
+) -> Result<GenericByteViewArray<T>, ArrowError> {
+    let new_views = take_native(array.views(), indices);
+    let new_nulls = take_nulls(array.nulls(), indices);
+    Ok(GenericByteViewArray::new(
+        new_views,
+        array.data_buffers().to_vec(),
+        new_nulls,
+    ))
+}
+
 /// `take` implementation for list arrays
 ///
 /// Calculates the index and indexed offset for the inner array,
@@ -1424,6 +1444,53 @@ mod tests {
         assert_eq!(result.as_ref(), &expected);
     }
 
+    fn _test_byte_view<T>()
+    where
+        T: ByteViewType,
+        str: AsRef<T::Native>,
+        T::Native: PartialEq,
+    {
+        let index = UInt32Array::from(vec![Some(3), None, Some(1), Some(3), Some(4), Some(2)]);
+        let array = {
+            // ["hello", "world", null, "large payload over 12 bytes", "lulu"]
+            let mut builder = GenericByteViewBuilder::<T>::new();
+            builder.append_value("hello");
+            builder.append_value("world");
+            builder.append_null();
+            builder.append_value("large payload over 12 bytes");
+            builder.append_value("lulu");
+            builder.finish()
+        };
+
+        let actual = take(&array, &index, None).unwrap();
+
+        assert_eq!(actual.len(), index.len());
+
+        let expected = {
+            // ["large payload over 12 bytes", null, "world", "large payload over 12 bytes", "lulu", null]
+            let mut builder = GenericByteViewBuilder::<T>::new();
+            builder.append_value("large payload over 12 bytes");
+            builder.append_null();
+            builder.append_value("world");
+            builder.append_value("large payload over 12 bytes");
+            builder.append_value("lulu");
+            builder.append_null();
+            builder.finish()
+        };
+
+        assert_eq!(actual.as_ref(), &expected);
+    }
+
+    #[test]
+    fn test_take_string_view() {
+        _test_byte_view::<StringViewType>()
+    }
+
+    #[test]
+    fn test_take_binary_view() {
+        _test_byte_view::<BinaryViewType>()
+    }
+
     macro_rules! test_take_list {
         ($offset_type:ty, $list_data_type:ident, $list_array_type:ident) => {{
             // Construct a value array, [[0,0,0], [-1,-2,-1], [], [2,3]]
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 4d101456f..838bef8ac 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -268,6 +268,7 @@ pub(crate) struct Request<'a> {
     builder: RequestBuilder,
     payload_sha256: Option<Vec<u8>>,
     use_session_creds: bool,
+    idempotent: bool,
 }
 
 impl<'a> Request<'a> {
@@ -285,6 +286,11 @@ impl<'a> Request<'a> {
         Self { builder, ..self }
     }
 
+    pub fn set_idempotent(mut self, idempotent: bool) -> Self {
+        self.idempotent = idempotent;
+        self
+    }
+
     pub async fn send(self) -> Result<Response, RequestError> {
         let credential = match self.use_session_creds {
             true => self.config.get_session_credential().await?,
@@ -298,7 +304,7 @@ impl<'a> Request<'a> {
         let path = self.path.as_ref();
         self.builder
             .with_aws_sigv4(credential.authorizer(), self.payload_sha256.as_deref())
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, self.idempotent)
             .await
             .context(RetrySnafu { path })
     }
@@ -360,6 +366,7 @@ impl S3Client {
             payload_sha256,
             config: &self.config,
             use_session_creds: true,
+            idempotent: false,
         }
     }
 
@@ -462,7 +469,7 @@ impl S3Client {
             .header(CONTENT_TYPE, "application/xml")
             .body(body)
             .with_aws_sigv4(credential.authorizer(), payload_sha256.as_deref())
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, false)
             .await
             .context(DeleteObjectsRequestSnafu {})?
             .bytes()
@@ -510,6 +517,7 @@ impl S3Client {
             config: &self.config,
             payload_sha256: None,
             use_session_creds: false,
+            idempotent: false,
         }
     }
 
@@ -522,7 +530,7 @@ impl S3Client {
             .request(Method::POST, url)
             .headers(self.config.encryption_headers.clone().into())
             .with_aws_sigv4(credential.authorizer(), None)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .context(CreateMultipartRequestSnafu)?
             .bytes()
@@ -547,6 +555,7 @@ impl S3Client {
         let response = self
             .put_request(path, data, false)
             .query(&[("partNumber", &part), ("uploadId", upload_id)])
+            .set_idempotent(true)
             .send()
             .await?;
 
@@ -582,7 +591,7 @@ impl S3Client {
             .query(&[("uploadId", upload_id)])
             .body(body)
             .with_aws_sigv4(credential.authorizer(), None)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .context(CompleteMultipartRequestSnafu)?;
 
diff --git a/object_store/src/aws/credential.rs b/object_store/src/aws/credential.rs
index 478e56dd0..a7d1a9772 100644
--- a/object_store/src/aws/credential.rs
+++ b/object_store/src/aws/credential.rs
@@ -517,7 +517,7 @@ async fn instance_creds(
     let token_result = client
         .request(Method::PUT, token_url)
         .header("X-aws-ec2-metadata-token-ttl-seconds", "600") // 10 minute TTL
-        .send_retry(retry_config)
+        .send_retry_with_idempotency(retry_config, true)
         .await;
 
     let token = match token_result {
@@ -607,7 +607,7 @@ async fn web_identity(
             ("Version", "2011-06-15"),
             ("WebIdentityToken", &token),
         ])
-        .send_retry(retry_config)
+        .send_retry_with_idempotency(retry_config, true)
         .await?
         .bytes()
         .await?;
diff --git a/object_store/src/aws/dynamo.rs b/object_store/src/aws/dynamo.rs
index 2390187e7..2e60bbad2 100644
--- a/object_store/src/aws/dynamo.rs
+++ b/object_store/src/aws/dynamo.rs
@@ -186,7 +186,11 @@ impl DynamoCommit {
         to: &Path,
     ) -> Result<()> {
         self.conditional_op(client, to, None, || async {
-            client.copy_request(from, to).send().await?;
+            client
+                .copy_request(from, to)
+                .set_idempotent(false)
+                .send()
+                .await?;
             Ok(())
         })
         .await
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 76d01d597..16af4d3b4 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -159,7 +159,7 @@ impl ObjectStore for AmazonS3 {
         }
 
         match (opts.mode, &self.client.config.conditional_put) {
-            (PutMode::Overwrite, _) => request.do_put().await,
+            (PutMode::Overwrite, _) => request.set_idempotent(true).do_put().await,
             (PutMode::Create | PutMode::Update(_), None) => Err(Error::NotImplemented),
             (PutMode::Create, Some(S3ConditionalPut::ETagMatch)) => {
                 match request.header(&IF_NONE_MATCH, "*").do_put().await {
@@ -268,7 +268,11 @@ impl ObjectStore for AmazonS3 {
     }
 
     async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
-        self.client.copy_request(from, to).send().await?;
+        self.client
+            .copy_request(from, to)
+            .set_idempotent(true)
+            .send()
+            .await?;
         Ok(())
     }
 
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index 5be6658be..0e6af50fb 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -172,6 +172,7 @@ struct PutRequest<'a> {
     path: &'a Path,
     config: &'a AzureConfig,
     builder: RequestBuilder,
+    idempotent: bool,
 }
 
 impl<'a> PutRequest<'a> {
@@ -185,12 +186,17 @@ impl<'a> PutRequest<'a> {
         Self { builder, ..self }
     }
 
+    fn set_idempotent(mut self, idempotent: bool) -> Self {
+        self.idempotent = idempotent;
+        self
+    }
+
     async fn send(self) -> Result<Response> {
         let credential = self.config.get_credential().await?;
         let response = self
             .builder
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, self.idempotent)
             .await
             .context(PutRequestSnafu {
                 path: self.path.as_ref(),
@@ -239,6 +245,7 @@ impl AzureClient {
             path,
             builder,
             config: &self.config,
+            idempotent: false,
         }
     }
 
@@ -247,7 +254,7 @@ impl AzureClient {
         let builder = self.put_request(path, bytes);
 
         let builder = match &opts.mode {
-            PutMode::Overwrite => builder,
+            PutMode::Overwrite => builder.set_idempotent(true),
             PutMode::Create => builder.header(&IF_NONE_MATCH, "*"),
             PutMode::Update(v) => {
                 let etag = v.e_tag.as_ref().context(MissingETagSnafu)?;
@@ -271,6 +278,7 @@ impl AzureClient {
 
         self.put_request(path, data)
             .query(&[("comp", "block"), ("blockid", &block_id)])
+            .set_idempotent(true)
             .send()
             .await?;
 
@@ -287,6 +295,7 @@ impl AzureClient {
         let response = self
             .put_request(path, BlockList { blocks }.to_xml().into())
             .query(&[("comp", "blocklist")])
+            .set_idempotent(true)
             .send()
             .await?;
 
@@ -340,7 +349,7 @@ impl AzureClient {
 
         builder
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .map_err(|err| err.error(STORE, from.to_string()))?;
 
@@ -373,7 +382,7 @@ impl AzureClient {
             .body(body)
             .query(&[("restype", "service"), ("comp", "userdelegationkey")])
             .with_azure_authorization(&credential, &self.config.account)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .context(DelegationKeyRequestSnafu)?
             .bytes()
diff --git a/object_store/src/azure/credential.rs b/object_store/src/azure/credential.rs
index 6dc3141b0..36845bd1d 100644
--- a/object_store/src/azure/credential.rs
+++ b/object_store/src/azure/credential.rs
@@ -615,7 +615,7 @@ impl TokenProvider for ClientSecretOAuthProvider {
                 ("scope", AZURE_STORAGE_SCOPE),
                 ("grant_type", "client_credentials"),
             ])
-            .send_retry(retry)
+            .send_retry_with_idempotency(retry, true)
             .await
             .context(TokenRequestSnafu)?
             .json()
@@ -797,7 +797,7 @@ impl TokenProvider for WorkloadIdentityOAuthProvider {
                 ("scope", AZURE_STORAGE_SCOPE),
                 ("grant_type", "client_credentials"),
             ])
-            .send_retry(retry)
+            .send_retry_with_idempotency(retry, true)
             .await
             .context(TokenRequestSnafu)?
             .json()
diff --git a/object_store/src/client/retry.rs b/object_store/src/client/retry.rs
index e4bb5c9e7..f3fa7153e 100644
--- a/object_store/src/client/retry.rs
+++ b/object_store/src/client/retry.rs
@@ -166,128 +166,83 @@ impl Default for RetryConfig {
     }
 }
 
-pub trait RetryExt {
-    /// Dispatch a request with the given retry configuration
-    ///
-    /// # Panic
-    ///
-    /// This will panic if the request body is a stream
-    fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<Response>>;
-}
-
-impl RetryExt for reqwest::RequestBuilder {
-    fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<Response>> {
-        let mut backoff = Backoff::new(&config.backoff);
-        let max_retries = config.max_retries;
-        let retry_timeout = config.retry_timeout;
-
-        let (client, req) = self.build_split();
-        let req = req.expect("request must be valid");
-
-        async move {
-            let mut retries = 0;
-            let now = Instant::now();
-
-            loop {
-                let s = req.try_clone().expect("request body must be cloneable");
-                match client.execute(s).await {
-                    Ok(r) => match r.error_for_status_ref() {
-                        Ok(_) if r.status().is_success() => return Ok(r),
-                        Ok(r) if r.status() == StatusCode::NOT_MODIFIED => {
-                            return Err(Error::Client {
+fn send_retry_impl(
+    builder: reqwest::RequestBuilder,
+    config: &RetryConfig,
+    is_idempotent: Option<bool>,
+) -> BoxFuture<'static, Result<Response>> {
+    let mut backoff = Backoff::new(&config.backoff);
+    let max_retries = config.max_retries;
+    let retry_timeout = config.retry_timeout;
+
+    let (client, req) = builder.build_split();
+    let req = req.expect("request must be valid");
+    let is_idempotent = is_idempotent.unwrap_or(req.method().is_safe());
+
+    async move {
+        let mut retries = 0;
+        let now = Instant::now();
+
+        loop {
+            let s = req.try_clone().expect("request body must be cloneable");
+            match client.execute(s).await {
+                Ok(r) => match r.error_for_status_ref() {
+                    Ok(_) if r.status().is_success() => return Ok(r),
+                    Ok(r) if r.status() == StatusCode::NOT_MODIFIED => {
+                        return Err(Error::Client {
+                            body: None,
+                            status: StatusCode::NOT_MODIFIED,
+                        })
+                    }
+                    Ok(r) => {
+                        let is_bare_redirect = r.status().is_redirection() && !r.headers().contains_key(LOCATION);
+                        return match is_bare_redirect {
+                            true => Err(Error::BareRedirect),
+                            // Not actually sure if this is reachable, but here for completeness
+                            false => Err(Error::Client {
                                 body: None,
-                                status: StatusCode::NOT_MODIFIED,
+                                status: r.status(),
                             })
                         }
-                        Ok(r) => {
-                            let is_bare_redirect = r.status().is_redirection() && !r.headers().contains_key(LOCATION);
-                            return match is_bare_redirect {
-                                true => Err(Error::BareRedirect),
-                                // Not actually sure if this is reachable, but here for completeness
-                                false => Err(Error::Client {
-                                    body: None,
-                                    status: r.status(),
-                                })
-                            }
-                        }
-                        Err(e) => {
-                            let status = r.status();
-                            if retries == max_retries
-                                || now.elapsed() > retry_timeout
-                                || !status.is_server_error() {
-
-                                return Err(match status.is_client_error() {
-                                    true => match r.text().await {
-                                        Ok(body) => {
-                                            Error::Client {
-                                                body: Some(body).filter(|b| !b.is_empty()),
-                                                status,
-                                            }
-                                        }
-                                        Err(e) => {
-                                            Error::Reqwest {
-                                                retries,
-                                                max_retries,
-                                                elapsed: now.elapsed(),
-                                                retry_timeout,
-                                                source: e,
-                                            }
+                    }
+                    Err(e) => {
+                        let status = r.status();
+                        if retries == max_retries
+                            || now.elapsed() > retry_timeout
+                            || !status.is_server_error() {
+
+                            return Err(match status.is_client_error() {
+                                true => match r.text().await {
+                                    Ok(body) => {
+                                        Error::Client {
+                                            body: Some(body).filter(|b| !b.is_empty()),
+                                            status,
                                         }
                                     }
-                                    false => Error::Reqwest {
-                                        retries,
-                                        max_retries,
-                                        elapsed: now.elapsed(),
-                                        retry_timeout,
-                                        source: e,
+                                    Err(e) => {
+                                        Error::Reqwest {
+                                            retries,
+                                            max_retries,
+                                            elapsed: now.elapsed(),
+                                            retry_timeout,
+                                            source: e,
+                                        }
                                     }
-                                });
-                            }
-
-                            let sleep = backoff.next();
-                            retries += 1;
-                            info!(
-                                "Encountered server error, backing off for {} seconds, retry {} of {}: {}",
-                                sleep.as_secs_f32(),
-                                retries,
-                                max_retries,
-                                e,
-                            );
-                            tokio::time::sleep(sleep).await;
-                        }
-                    },
-                    Err(e) =>
-                    {
-                        let mut do_retry = false;
-                        if e.is_connect() || ( req.method().is_safe() && e.is_timeout()) {
-                            do_retry = true
-                        } else {
-                            let mut source = e.source();
-                            while let Some(e) = source {
-                                if let Some(e) = e.downcast_ref::<hyper::Error>() {
-                                    do_retry = e.is_closed() || e.is_incomplete_message();
-                                    break
                                 }
-                                source = e.source();
-                            }
+                                false => Error::Reqwest {
+                                    retries,
+                                    max_retries,
+                                    elapsed: now.elapsed(),
+                                    retry_timeout,
+                                    source: e,
+                                }
+                            });
                         }
 
-                        if retries == max_retries
-                            || now.elapsed() > retry_timeout
-                            || !do_retry {
-
-                            return Err(Error::Reqwest {
-                                retries,
-                                max_retries,
-                                elapsed: now.elapsed(),
-                                retry_timeout,
-                                source: e,
-                            })
-                        }
                         let sleep = backoff.next();
                         retries += 1;
                         info!(
-                            "Encountered transport error backing off for {} seconds, retry {} of {}: {}", 
+                            "Encountered server error, backing off for {} seconds, retry {} of {}: {}",
                             sleep.as_secs_f32(),
                             retries,
                             max_retries,
@@ -295,10 +250,102 @@ impl RetryExt for reqwest::RequestBuilder {
                         );
                         tokio::time::sleep(sleep).await;
                     }
+                },
+                Err(e) =>
+                {
+                    let mut do_retry = false;
+                    if e.is_connect()
+                        || e.is_body()
+                        || (e.is_request() && !e.is_timeout())
+                        || (is_idempotent && e.is_timeout()) {
+                        do_retry = true
+                    } else {
+                        let mut source = e.source();
+                        while let Some(e) = source {
+                            if let Some(e) = e.downcast_ref::<hyper::Error>() {
+                                do_retry = e.is_closed()
+                                    || e.is_incomplete_message()
+                                    || e.is_body_write_aborted()
+                                    || (is_idempotent && e.is_timeout());
+                                break
+                            }
+                            if let Some(e) = e.downcast_ref::<std::io::Error>() {
+                                if e.kind() == std::io::ErrorKind::TimedOut {
+                                    do_retry = is_idempotent;
+                                } else {
+                                    do_retry = matches!(
+                                        e.kind(),
+                                        std::io::ErrorKind::ConnectionReset
+                                        | std::io::ErrorKind::ConnectionAborted
+                                        | std::io::ErrorKind::BrokenPipe
+                                        | std::io::ErrorKind::UnexpectedEof
+                                    );
+                                }
+                                break;
+                            }
+                            source = e.source();
+                        }
+                    }
+
+                    if retries == max_retries
+                        || now.elapsed() > retry_timeout
+                        || !do_retry {
+
+                        return Err(Error::Reqwest {
+                            retries,
+                            max_retries,
+                            elapsed: now.elapsed(),
+                            retry_timeout,
+                            source: e,
+                        })
+                    }
+                    let sleep = backoff.next();
+                    retries += 1;
+                    info!(
+                        "Encountered transport error backing off for {} seconds, retry {} of {}: {}",
+                        sleep.as_secs_f32(),
+                        retries,
+                        max_retries,
+                        e,
+                    );
+                    tokio::time::sleep(sleep).await;
                 }
             }
         }
-        .boxed()
+    }
+    .boxed()
+}
+
+pub trait RetryExt {
+    /// Dispatch a request with the given retry configuration
+    ///
+    /// # Panic
+    ///
+    /// This will panic if the request body is a stream
+    fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<Response>>;
+
+    /// Dispatch a request with the given retry configuration and idempotency
+    ///
+    /// # Panic
+    ///
+    /// This will panic if the request body is a stream
+    fn send_retry_with_idempotency(
+        self,
+        config: &RetryConfig,
+        is_idempotent: bool,
+    ) -> BoxFuture<'static, Result<Response>>;
+}
+
+impl RetryExt for reqwest::RequestBuilder {
+    fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<Response>> {
+        send_retry_impl(self, config, None)
+    }
+    fn send_retry_with_idempotency(
+        self,
+        config: &RetryConfig,
+        is_idempotent: bool,
+    ) -> BoxFuture<'static, Result<Response>> {
+        send_retry_impl(self, config, Some(is_idempotent))
     }
 }
 
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index 3762915b0..17404f9d5 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -173,6 +173,7 @@ pub struct PutRequest<'a> {
     path: &'a Path,
     config: &'a GoogleCloudStorageConfig,
     builder: RequestBuilder,
+    idempotent: bool,
 }
 
 impl<'a> PutRequest<'a> {
@@ -186,12 +187,17 @@ impl<'a> PutRequest<'a> {
         Self { builder, ..self }
     }
 
+    fn set_idempotent(mut self, idempotent: bool) -> Self {
+        self.idempotent = idempotent;
+        self
+    }
+
     async fn send(self) -> Result<PutResult> {
         let credential = self.config.credentials.get_credential().await?;
         let response = self
             .builder
             .bearer_auth(&credential.bearer)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, self.idempotent)
             .await
             .context(PutRequestSnafu {
                 path: self.path.as_ref(),
@@ -281,7 +287,7 @@ impl GoogleCloudStorageClient {
             .post(&url)
             .bearer_auth(&credential.bearer)
             .json(&body)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .context(SignBlobRequestSnafu)?;
 
@@ -329,6 +335,7 @@ impl GoogleCloudStorageClient {
             path,
             builder,
             config: &self.config,
+            idempotent: false,
         }
     }
 
@@ -336,7 +343,7 @@ impl GoogleCloudStorageClient {
         let builder = self.put_request(path, data);
 
         let builder = match &opts.mode {
-            PutMode::Overwrite => builder,
+            PutMode::Overwrite => builder.set_idempotent(true),
             PutMode::Create => builder.header(&VERSION_MATCH, "0"),
             PutMode::Update(v) => {
                 let etag = v.version.as_ref().context(MissingVersionSnafu)?;
@@ -366,7 +373,12 @@ impl GoogleCloudStorageClient {
             ("partNumber", &format!("{}", part_idx + 1)),
             ("uploadId", upload_id),
         ];
-        let result = self.put_request(path, data).query(query).send().await?;
+        let result = self
+            .put_request(path, data)
+            .query(query)
+            .set_idempotent(true)
+            .send()
+            .await?;
 
         Ok(PartId {
             content_id: result.e_tag.unwrap(),
@@ -391,7 +403,7 @@ impl GoogleCloudStorageClient {
             .header(header::CONTENT_TYPE, content_type)
             .header(header::CONTENT_LENGTH, "0")
             .query(&[("uploads", "")])
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .context(PutRequestSnafu {
                 path: path.as_ref(),
@@ -432,7 +444,11 @@ impl GoogleCloudStorageClient {
     ) -> Result<PutResult> {
         if completed_parts.is_empty() {
             // GCS doesn't allow empty multipart uploads
-            let result = self.put_request(path, Default::default()).send().await?;
+            let result = self
+                .put_request(path, Default::default())
+                .set_idempotent(true)
+                .send()
+                .await?;
             self.multipart_cleanup(path, multipart_id).await?;
             return Ok(result);
         }
@@ -456,7 +472,7 @@ impl GoogleCloudStorageClient {
             .bearer_auth(&credential.bearer)
             .query(&[("uploadId", upload_id)])
             .body(data)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, true)
             .await
             .context(CompleteMultipartRequestSnafu)?;
 
@@ -515,7 +531,7 @@ impl GoogleCloudStorageClient {
             // Needed if reqwest is compiled with native-tls instead of rustls-tls
             // See https://github.com/apache/arrow-rs/pull/3921
             .header(header::CONTENT_LENGTH, 0)
-            .send_retry(&self.config.retry_config)
+            .send_retry_with_idempotency(&self.config.retry_config, !if_not_exists)
             .await
             .map_err(|err| match err.status() {
                 Some(StatusCode::PRECONDITION_FAILED) => crate::Error::AlreadyExists {
diff --git a/object_store/src/gcp/credential.rs b/object_store/src/gcp/credential.rs
index fcd516a1b..158716ce4 100644
--- a/object_store/src/gcp/credential.rs
+++ b/object_store/src/gcp/credential.rs
@@ -623,7 +623,7 @@ impl TokenProvider for AuthorizedUserCredentials {
                 ("client_secret", &self.client_secret),
                 ("refresh_token", &self.refresh_token),
             ])
-            .send_retry(retry)
+            .send_retry_with_idempotency(retry, true)
             .await
             .context(TokenRequestSnafu)?
             .json::<TokenResponse>()
@@ -709,12 +709,12 @@ impl GCSAuthorizer {
     /// Canonicalizes query parameters into the GCP canonical form
     /// form like:
     ///```plaintext
-    ///HTTP_VERB  
-    ///PATH_TO_RESOURCE  
-    ///CANONICAL_QUERY_STRING  
-    ///CANONICAL_HEADERS  
+    ///HTTP_VERB
+    ///PATH_TO_RESOURCE
+    ///CANONICAL_QUERY_STRING
+    ///CANONICAL_HEADERS
     ///
-    ///SIGNED_HEADERS  
+    ///SIGNED_HEADERS
     ///PAYLOAD
     ///```
     ///
@@ -780,9 +780,9 @@ impl GCSAuthorizer {
     ///construct the string to sign
     ///form like:
     ///```plaintext
-    ///SIGNING_ALGORITHM  
-    ///ACTIVE_DATETIME  
-    ///CREDENTIAL_SCOPE  
+    ///SIGNING_ALGORITHM
+    ///ACTIVE_DATETIME
+    ///CREDENTIAL_SCOPE
     ///HASHED_CANONICAL_REQUEST
     ///```
     ///`ACTIVE_DATETIME` format:`YYYYMMDD'T'HHMMSS'Z'`
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index 8700775fb..fdc8751c1 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -189,7 +189,7 @@ impl Client {
             .client
             .request(method, url)
             .header("Depth", depth)
-            .send_retry(&self.retry_config)
+            .send_retry_with_idempotency(&self.retry_config, true)
             .await;
 
         let response = match result {
diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
index 80a554026..c8a9f82c3 100644
--- a/parquet/src/arrow/async_reader/mod.rs
+++ b/parquet/src/arrow/async_reader/mod.rs
@@ -1857,4 +1857,92 @@ mod tests {
             assert_eq!(total_rows, expected);
         }
     }
+
+    #[tokio::test]
+    async fn test_row_filter_nested() {
+        let a = StringArray::from_iter_values(["a", "b", "b", "b", "c", "c"]);
+        let b = StructArray::from(vec![
+            (
+                Arc::new(Field::new("aa", DataType::Utf8, true)),
+                Arc::new(StringArray::from(vec!["a", "b", "b", "b", "c", "c"])) as ArrayRef,
+            ),
+            (
+                Arc::new(Field::new("bb", DataType::Utf8, true)),
+                Arc::new(StringArray::from(vec!["1", "2", "3", "4", "5", "6"])) as ArrayRef,
+            ),
+        ]);
+        let c = Int32Array::from_iter(0..6);
+        let data = RecordBatch::try_from_iter([
+            ("a", Arc::new(a) as ArrayRef),
+            ("b", Arc::new(b) as ArrayRef),
+            ("c", Arc::new(c) as ArrayRef),
+        ])
+        .unwrap();
+
+        let mut buf = Vec::with_capacity(1024);
+        let mut writer = ArrowWriter::try_new(&mut buf, data.schema(), None).unwrap();
+        writer.write(&data).unwrap();
+        writer.close().unwrap();
+
+        let data: Bytes = buf.into();
+        let metadata = parse_metadata(&data).unwrap();
+        let parquet_schema = metadata.file_metadata().schema_descr_ptr();
+
+        let test = TestReader {
+            data,
+            metadata: Arc::new(metadata),
+            requests: Default::default(),
+        };
+        let requests = test.requests.clone();
+
+        let a_scalar = StringArray::from_iter_values(["b"]);
+        let a_filter = ArrowPredicateFn::new(
+            ProjectionMask::leaves(&parquet_schema, vec![0]),
+            move |batch| eq(batch.column(0), &Scalar::new(&a_scalar)),
+        );
+
+        let b_scalar = StringArray::from_iter_values(["4"]);
+        let b_filter = ArrowPredicateFn::new(
+            ProjectionMask::leaves(&parquet_schema, vec![2]),
+            move |batch| {
+                // Filter on the second element of the struct.
+                let struct_array = batch
+                    .column(0)
+                    .as_any()
+                    .downcast_ref::<StructArray>()
+                    .unwrap();
+                eq(struct_array.column(0), &Scalar::new(&b_scalar))
+            },
+        );
+
+        let filter = RowFilter::new(vec![Box::new(a_filter), Box::new(b_filter)]);
+
+        let mask = ProjectionMask::leaves(&parquet_schema, vec![0, 3]);
+        let stream = ParquetRecordBatchStreamBuilder::new(test)
+            .await
+            .unwrap()
+            .with_projection(mask.clone())
+            .with_batch_size(1024)
+            .with_row_filter(filter)
+            .build()
+            .unwrap();
+
+        let batches: Vec<_> = stream.try_collect().await.unwrap();
+        assert_eq!(batches.len(), 1);
+
+        let batch = &batches[0];
+        assert_eq!(batch.num_rows(), 1);
+        assert_eq!(batch.num_columns(), 2);
+
+        let col = batch.column(0);
+        let val = col.as_any().downcast_ref::<StringArray>().unwrap().value(0);
+        assert_eq!(val, "b");
+
+        let col = batch.column(1);
+        let val = col.as_any().downcast_ref::<Int32Array>().unwrap().value(0);
+        assert_eq!(val, 3);
+
+        // Should only have made 3 requests
+        assert_eq!(requests.lock().unwrap().len(), 3);
+    }
 }


Test patch:
diff --git a/arrow-flight/tests/encode_decode.rs b/arrow-flight/tests/encode_decode.rs
index 224b12500..0185fa77f 100644
--- a/arrow-flight/tests/encode_decode.rs
+++ b/arrow-flight/tests/encode_decode.rs
@@ -20,7 +20,10 @@
 use std::{collections::HashMap, sync::Arc};
 
 use arrow_array::types::Int32Type;
-use arrow_array::{ArrayRef, DictionaryArray, Float64Array, RecordBatch, UInt8Array};
+use arrow_array::{
+    ArrayRef, BinaryViewArray, DictionaryArray, Float64Array, RecordBatch, StringViewArray,
+    UInt8Array,
+};
 use arrow_cast::pretty::pretty_format_batches;
 use arrow_flight::flight_descriptor::DescriptorType;
 use arrow_flight::FlightDescriptor;
@@ -111,6 +114,22 @@ async fn test_dictionary_many() {
     .await;
 }
 
+#[tokio::test]
+async fn test_view_types_one() {
+    roundtrip(vec![make_view_batches(5)]).await;
+}
+
+#[tokio::test]
+async fn test_view_types_many() {
+    roundtrip(vec![
+        make_view_batches(5),
+        make_view_batches(9),
+        make_view_batches(5),
+        make_view_batches(5),
+    ])
+    .await;
+}
+
 #[tokio::test]
 async fn test_zero_batches_no_schema() {
     let stream = FlightDataEncoderBuilder::default().build(futures::stream::iter(vec![]));
@@ -450,8 +469,43 @@ fn make_dictionary_batch(num_rows: usize) -> RecordBatch {
     RecordBatch::try_from_iter(vec![("a", Arc::new(a) as ArrayRef)]).unwrap()
 }
 
+fn make_view_batches(num_rows: usize) -> RecordBatch {
+    const LONG_TEST_STRING: &str =
+        "This is a long string to make sure binary view array handles it";
+    let schema = Schema::new(vec![
+        Field::new("field1", DataType::BinaryView, true),
+        Field::new("field2", DataType::Utf8View, true),
+    ]);
+
+    let string_view_values: Vec<Option<&str>> = (0..num_rows)
+        .map(|i| match i % 3 {
+            0 => None,
+            1 => Some("foo"),
+            2 => Some(LONG_TEST_STRING),
+            _ => unreachable!(),
+        })
+        .collect();
+
+    let bin_view_values: Vec<Option<&[u8]>> = (0..num_rows)
+        .map(|i| match i % 3 {
+            0 => None,
+            1 => Some("bar".as_bytes()),
+            2 => Some(LONG_TEST_STRING.as_bytes()),
+            _ => unreachable!(),
+        })
+        .collect();
+
+    let binary_array = BinaryViewArray::from_iter(bin_view_values);
+    let utf8_array = StringViewArray::from_iter(string_view_values);
+    RecordBatch::try_new(
+        Arc::new(schema.clone()),
+        vec![Arc::new(binary_array), Arc::new(utf8_array)],
+    )
+    .unwrap()
+}
+
 /// Encodes input as a FlightData stream, and then decodes it using
-/// FlightRecordBatchStream and valides the decoded record batches
+/// FlightRecordBatchStream and validates the decoded record batches
 /// match the input.
 async fn roundtrip(input: Vec<RecordBatch>) {
     let expected_output = input.clone();
@@ -459,7 +513,7 @@ async fn roundtrip(input: Vec<RecordBatch>) {
 }
 
 /// Encodes input as a FlightData stream, and then decodes it using
-/// FlightRecordBatchStream and valides the decoded record batches
+/// FlightRecordBatchStream and validates the decoded record batches
 /// match the expected input.
 ///
 /// When <https://github.com/apache/arrow-rs/issues/3389> is resolved,