Repository: apache/arrow-rs
Pull request #5919 (2024-06-19 14:40:51)

Add `MultipartUpload` blanket implementation for `Box<W>`
# Which issue does this PR close?

None

# Rationale for this change
 
It makes API more pleasant to work with as usually we have traits inside the box

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/README.md b/README.md
index ac676f7df..335ae2415 100644
--- a/README.md
+++ b/README.md
@@ -79,15 +79,19 @@ versions approximately every 2 months.
 
 ## Related Projects
 
-There are two related crates in different repositories
+There are several related crates in different repositories
 
-| Crate          | Description                             | Documentation                 |
-| -------------- | --------------------------------------- | ----------------------------- |
-| [`datafusion`] | In-memory query engine with SQL support | [(README)][datafusion-readme] |
-| [`ballista`]   | Distributed query execution             | [(README)][ballista-readme]   |
+| Crate                    | Description                                 | Documentation                           |
+| ------------------------ | ------------------------------------------- | --------------------------------------- |
+| [`datafusion`]           | In-memory query engine with SQL support     | [(README)][datafusion-readme]           |
+| [`ballista`]             | Distributed query execution                 | [(README)][ballista-readme]             |
+| [`object_store_opendal`] | Use [`opendal`] as [`object_store`] backend | [(README)][object_store_opendal-readme] |
 
 [`datafusion`]: https://crates.io/crates/datafusion
 [`ballista`]: https://crates.io/crates/ballista
+[`object_store_opendal`]: https://crates.io/crates/object_store_opendal
+[`opendal`]: https://crates.io/crates/opendal
+[object_store_opendal-readme]: https://github.com/apache/opendal/blob/main/integrations/object_store/README.md
 
 Collectively, these crates support a wider array of functionality for analytic computations in Rust.
 
diff --git a/arrow-array/src/ffi.rs b/arrow-array/src/ffi.rs
index 088a0a6ab..fe755b91f 100644
--- a/arrow-array/src/ffi.rs
+++ b/arrow-array/src/ffi.rs
@@ -1225,6 +1225,7 @@ mod tests_from_ffi {
     use std::sync::Arc;
 
     use arrow_buffer::{bit_util, buffer::Buffer, MutableBuffer, OffsetBuffer};
+    use arrow_data::transform::MutableArrayData;
     use arrow_data::ArrayData;
     use arrow_schema::{DataType, Field};
 
@@ -1234,6 +1235,7 @@ mod tests_from_ffi {
             Int32Array, Int64Array, StringArray, StructArray, UInt32Array, UInt64Array,
         },
         ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema},
+        make_array, ArrayRef,
     };
 
     use super::{ImportedArrowArray, Result};
@@ -1458,4 +1460,58 @@ mod tests_from_ffi {
 
         test_round_trip(&imported_array.consume()?)
     }
+
+    fn roundtrip_string_array(array: StringArray) -> StringArray {
+        let data = array.into_data();
+
+        let array = FFI_ArrowArray::new(&data);
+        let schema = FFI_ArrowSchema::try_from(data.data_type()).unwrap();
+
+        let array = unsafe { from_ffi(array, &schema) }.unwrap();
+        StringArray::from(array)
+    }
+
+    fn extend_array(array: &dyn Array) -> ArrayRef {
+        let len = array.len();
+        let data = array.to_data();
+
+        let mut mutable = MutableArrayData::new(vec![&data], false, len);
+        mutable.extend(0, 0, len);
+        make_array(mutable.freeze())
+    }
+
+    #[test]
+    fn test_extend_imported_string_slice() {
+        let mut strings = vec![];
+
+        for i in 0..1000 {
+            strings.push(format!("string: {}", i));
+        }
+
+        let string_array = StringArray::from(strings);
+
+        let imported = roundtrip_string_array(string_array.clone());
+        assert_eq!(imported.len(), 1000);
+        assert_eq!(imported.value(0), "string: 0");
+        assert_eq!(imported.value(499), "string: 499");
+
+        let copied = extend_array(&imported);
+        assert_eq!(
+            copied.as_any().downcast_ref::<StringArray>().unwrap(),
+            &imported
+        );
+
+        let slice = string_array.slice(500, 500);
+
+        let imported = roundtrip_string_array(slice);
+        assert_eq!(imported.len(), 500);
+        assert_eq!(imported.value(0), "string: 500");
+        assert_eq!(imported.value(499), "string: 999");
+
+        let copied = extend_array(&imported);
+        assert_eq!(
+            copied.as_any().downcast_ref::<StringArray>().unwrap(),
+            &imported
+        );
+    }
 }
diff --git a/arrow-buffer/src/buffer/immutable.rs b/arrow-buffer/src/buffer/immutable.rs
index f26cde05b..2c743842f 100644
--- a/arrow-buffer/src/buffer/immutable.rs
+++ b/arrow-buffer/src/buffer/immutable.rs
@@ -71,6 +71,19 @@ impl Buffer {
         }
     }
 
+    /// Returns the offset, in bytes, of `Self::ptr` to `Self::data`
+    ///
+    /// self.ptr and self.data can be different after slicing or advancing the buffer.
+    pub fn ptr_offset(&self) -> usize {
+        // Safety: `ptr` is always in bounds of `data`.
+        unsafe { self.ptr.offset_from(self.data.ptr().as_ptr()) as usize }
+    }
+
+    /// Returns the pointer to the start of the buffer without the offset.
+    pub fn data_ptr(&self) -> NonNull<u8> {
+        self.data.ptr()
+    }
+
     /// Create a [`Buffer`] from the provided [`Vec`] without copying
     #[inline]
     pub fn from_vec<T: ArrowNativeType>(vec: Vec<T>) -> Self {
diff --git a/arrow-cast/src/cast/map.rs b/arrow-cast/src/cast/map.rs
new file mode 100644
index 000000000..d62a9519b
--- /dev/null
+++ b/arrow-cast/src/cast/map.rs
@@ -0,0 +1,74 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::cast::*;
+
+/// Helper function that takes a map container and casts the inner datatype.
+pub(crate) fn cast_map_values(
+    from: &MapArray,
+    to_data_type: &DataType,
+    cast_options: &CastOptions,
+    to_ordered: bool,
+) -> Result<ArrayRef, ArrowError> {
+    let entries_field = if let DataType::Map(entries_field, _) = to_data_type {
+        entries_field
+    } else {
+        return Err(ArrowError::CastError(
+            "Internal Error: to_data_type is not a map type.".to_string(),
+        ));
+    };
+
+    let key_field = key_field(entries_field).ok_or(ArrowError::CastError(
+        "map is missing key field".to_string(),
+    ))?;
+    let value_field = value_field(entries_field).ok_or(ArrowError::CastError(
+        "map is missing value field".to_string(),
+    ))?;
+
+    let key_array = cast_with_options(from.keys(), key_field.data_type(), cast_options)?;
+    let value_array = cast_with_options(from.values(), value_field.data_type(), cast_options)?;
+
+    Ok(Arc::new(MapArray::new(
+        entries_field.clone(),
+        from.offsets().clone(),
+        StructArray::new(
+            Fields::from(vec![key_field, value_field]),
+            vec![key_array, value_array],
+            from.entries().nulls().cloned(),
+        ),
+        from.nulls().cloned(),
+        to_ordered,
+    )))
+}
+
+/// Gets the key field from the entries of a map.  For all other types returns None.
+pub(crate) fn key_field(entries_field: &FieldRef) -> Option<FieldRef> {
+    if let DataType::Struct(fields) = entries_field.data_type() {
+        fields.first().cloned()
+    } else {
+        None
+    }
+}
+
+/// Gets the value field from the entries of a map.  For all other types returns None.
+pub(crate) fn value_field(entries_field: &FieldRef) -> Option<FieldRef> {
+    if let DataType::Struct(fields) = entries_field.data_type() {
+        fields.get(1).cloned()
+    } else {
+        None
+    }
+}
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index 55f2ed728..7a6e1a31b 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -40,10 +40,12 @@
 mod decimal;
 mod dictionary;
 mod list;
+mod map;
 mod string;
 use crate::cast::decimal::*;
 use crate::cast::dictionary::*;
 use crate::cast::list::*;
+use crate::cast::map::*;
 use crate::cast::string::*;
 
 use arrow_buffer::IntervalMonthDayNano;
@@ -159,6 +161,12 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
             can_cast_types(from_type, list_to.data_type())},
         (FixedSizeList(list_from,size), _) if *size == 1 => {
             can_cast_types(list_from.data_type(), to_type)},
+        (Map(from_entries,ordered_from), Map(to_entries, ordered_to)) if ordered_from == ordered_to =>
+            match (key_field(from_entries), key_field(to_entries), value_field(from_entries), value_field(to_entries)) {
+                (Some(from_key), Some(to_key), Some(from_value), Some(to_value)) =>
+                    can_cast_types(from_key.data_type(), to_key.data_type()) && can_cast_types(from_value.data_type(), to_value.data_type()),
+                _ => false
+            },
         // cast one decimal type to another decimal type
         (Decimal128(_, _), Decimal128(_, _)) => true,
         (Decimal256(_, _), Decimal256(_, _)) => true,
@@ -802,6 +810,9 @@ pub fn cast_with_options(
         (FixedSizeList(_, size), _) if *size == 1 => {
             cast_single_element_fixed_size_list_to_values(array, to_type, cast_options)
         }
+        (Map(_, ordered1), Map(_, ordered2)) if ordered1 == ordered2 => {
+            cast_map_values(array.as_map(), to_type, cast_options, ordered1.to_owned())
+        }
         (Decimal128(_, s1), Decimal128(p2, s2)) => {
             cast_decimal_to_decimal_same_type::<Decimal128Type>(
                 array.as_primitive(),
@@ -7361,6 +7372,240 @@ mod tests {
         FixedSizeListArray::from(list_data)
     }
 
+    #[test]
+    fn test_cast_map_dont_allow_change_of_order() {
+        let string_builder = StringBuilder::new();
+        let value_builder = StringBuilder::new();
+        let mut builder = MapBuilder::new(
+            Some(MapFieldNames {
+                entry: "entries".to_string(),
+                key: "key".to_string(),
+                value: "value".to_string(),
+            }),
+            string_builder,
+            value_builder,
+        );
+
+        builder.keys().append_value("0");
+        builder.values().append_value("test_val_1");
+        builder.append(true).unwrap();
+        builder.keys().append_value("1");
+        builder.values().append_value("test_val_2");
+        builder.append(true).unwrap();
+
+        // map builder returns unsorted map by default
+        let array = builder.finish();
+
+        let new_ordered = true;
+        let new_type = DataType::Map(
+            Arc::new(Field::new(
+                "entries",
+                DataType::Struct(
+                    vec![
+                        Field::new("key", DataType::Utf8, false),
+                        Field::new("value", DataType::Utf8, false),
+                    ]
+                    .into(),
+                ),
+                false,
+            )),
+            new_ordered,
+        );
+
+        let new_array_result = cast(&array, &new_type.clone());
+        assert!(!can_cast_types(array.data_type(), &new_type));
+        assert!(
+            matches!(new_array_result, Err(ArrowError::CastError(t)) if t == r#"Casting from Map(Field { name: "entries", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false) to Map(Field { name: "entries", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, true) not supported"#)
+        );
+    }
+
+    #[test]
+    fn test_cast_map_dont_allow_when_container_cant_cast() {
+        let string_builder = StringBuilder::new();
+        let value_builder = IntervalDayTimeArray::builder(2);
+        let mut builder = MapBuilder::new(
+            Some(MapFieldNames {
+                entry: "entries".to_string(),
+                key: "key".to_string(),
+                value: "value".to_string(),
+            }),
+            string_builder,
+            value_builder,
+        );
+
+        builder.keys().append_value("0");
+        builder.values().append_value(IntervalDayTime::new(1, 1));
+        builder.append(true).unwrap();
+        builder.keys().append_value("1");
+        builder.values().append_value(IntervalDayTime::new(2, 2));
+        builder.append(true).unwrap();
+
+        // map builder returns unsorted map by default
+        let array = builder.finish();
+
+        let new_ordered = true;
+        let new_type = DataType::Map(
+            Arc::new(Field::new(
+                "entries",
+                DataType::Struct(
+                    vec![
+                        Field::new("key", DataType::Utf8, false),
+                        Field::new("value", DataType::Duration(TimeUnit::Second), false),
+                    ]
+                    .into(),
+                ),
+                false,
+            )),
+            new_ordered,
+        );
+
+        let new_array_result = cast(&array, &new_type.clone());
+        assert!(!can_cast_types(array.data_type(), &new_type));
+        assert!(
+            matches!(new_array_result, Err(ArrowError::CastError(t)) if t == r#"Casting from Map(Field { name: "entries", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Interval(DayTime), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false) to Map(Field { name: "entries", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Duration(Second), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, true) not supported"#)
+        );
+    }
+
+    #[test]
+    fn test_cast_map_field_names() {
+        let string_builder = StringBuilder::new();
+        let value_builder = StringBuilder::new();
+        let mut builder = MapBuilder::new(
+            Some(MapFieldNames {
+                entry: "entries".to_string(),
+                key: "key".to_string(),
+                value: "value".to_string(),
+            }),
+            string_builder,
+            value_builder,
+        );
+
+        builder.keys().append_value("0");
+        builder.values().append_value("test_val_1");
+        builder.append(true).unwrap();
+        builder.keys().append_value("1");
+        builder.values().append_value("test_val_2");
+        builder.append(true).unwrap();
+        builder.append(false).unwrap();
+
+        let array = builder.finish();
+
+        let new_type = DataType::Map(
+            Arc::new(Field::new(
+                "entries_new",
+                DataType::Struct(
+                    vec![
+                        Field::new("key_new", DataType::Utf8, false),
+                        Field::new("value_values", DataType::Utf8, false),
+                    ]
+                    .into(),
+                ),
+                false,
+            )),
+            false,
+        );
+
+        assert_ne!(new_type, array.data_type().clone());
+
+        let new_array = cast(&array, &new_type.clone()).unwrap();
+        assert_eq!(new_type, new_array.data_type().clone());
+        let map_array = new_array.as_map();
+
+        assert_ne!(new_type, array.data_type().clone());
+        assert_eq!(new_type, map_array.data_type().clone());
+
+        let key_string = map_array
+            .keys()
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap()
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+        assert_eq!(&key_string, &vec!["0", "1"]);
+
+        let values_string_array = cast(map_array.values(), &DataType::Utf8).unwrap();
+        let values_string = values_string_array
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap()
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+        assert_eq!(&values_string, &vec!["test_val_1", "test_val_2"]);
+
+        assert_eq!(
+            map_array.nulls(),
+            Some(&NullBuffer::from(vec![true, true, false]))
+        );
+    }
+
+    #[test]
+    fn test_cast_map_contained_values() {
+        let string_builder = StringBuilder::new();
+        let value_builder = Int8Builder::new();
+        let mut builder = MapBuilder::new(
+            Some(MapFieldNames {
+                entry: "entries".to_string(),
+                key: "key".to_string(),
+                value: "value".to_string(),
+            }),
+            string_builder,
+            value_builder,
+        );
+
+        builder.keys().append_value("0");
+        builder.values().append_value(44);
+        builder.append(true).unwrap();
+        builder.keys().append_value("1");
+        builder.values().append_value(22);
+        builder.append(true).unwrap();
+
+        let array = builder.finish();
+
+        let new_type = DataType::Map(
+            Arc::new(Field::new(
+                "entries",
+                DataType::Struct(
+                    vec![
+                        Field::new("key", DataType::Utf8, false),
+                        Field::new("value", DataType::Utf8, false),
+                    ]
+                    .into(),
+                ),
+                false,
+            )),
+            false,
+        );
+
+        let new_array = cast(&array, &new_type.clone()).unwrap();
+        assert_eq!(new_type, new_array.data_type().clone());
+        let map_array = new_array.as_map();
+
+        assert_ne!(new_type, array.data_type().clone());
+        assert_eq!(new_type, map_array.data_type().clone());
+
+        let key_string = map_array
+            .keys()
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap()
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+        assert_eq!(&key_string, &vec!["0", "1"]);
+
+        let values_string_array = cast(map_array.values(), &DataType::Utf8).unwrap();
+        let values_string = values_string_array
+            .as_any()
+            .downcast_ref::<StringArray>()
+            .unwrap()
+            .into_iter()
+            .flatten()
+            .collect::<Vec<_>>();
+        assert_eq!(&values_string, &vec!["44", "22"]);
+    }
+
     #[test]
     fn test_utf8_cast_offsets() {
         // test if offset of the array is taken into account during cast
diff --git a/arrow-data/src/ffi.rs b/arrow-data/src/ffi.rs
index 589f7dac6..7fe87d4ae 100644
--- a/arrow-data/src/ffi.rs
+++ b/arrow-data/src/ffi.rs
@@ -131,6 +131,37 @@ impl FFI_ArrowArray {
             data.buffers().iter().map(|b| Some(b.clone())).collect()
         };
 
+        // Handle buffer offset for offset buffer.
+        let offset_offset = match data.data_type() {
+            DataType::Utf8 | DataType::Binary => {
+                // Offset buffer is possible a slice of the buffer.
+                // If we use slice pointer as exported buffer pointer, it will cause
+                // the consumer to calculate incorrect length of data buffer (buffer 1).
+                // We need to get the offset of the offset buffer and fill it in
+                // the `FFI_ArrowArray` offset field.
+                Some(data.buffers()[0].ptr_offset() / std::mem::size_of::<i32>())
+            }
+            DataType::LargeUtf8 | DataType::LargeBinary => {
+                // Offset buffer is possible a slice of the buffer.
+                // If we use slice pointer as exported buffer pointer, it will cause
+                // the consumer to calculate incorrect length of data buffer (buffer 1).
+                // We need to get the offset of the offset buffer and fill it in
+                // the `FFI_ArrowArray` offset field.
+                Some(data.buffers()[0].ptr_offset() / std::mem::size_of::<i64>())
+            }
+            _ => None,
+        };
+
+        let offset = if let Some(offset) = offset_offset {
+            if data.offset() != 0 {
+                // TODO: Adjust for data offset
+                panic!("The ArrayData of a slice offset buffer should not have offset");
+            }
+            offset
+        } else {
+            data.offset()
+        };
+
         // `n_buffers` is the number of buffers by the spec.
         let n_buffers = {
             data_layout.buffers.len() + {
@@ -143,9 +174,25 @@ impl FFI_ArrowArray {
 
         let buffers_ptr = buffers
             .iter()
-            .flat_map(|maybe_buffer| match maybe_buffer {
-                // note that `raw_data` takes into account the buffer's offset
-                Some(b) => Some(b.as_ptr() as *const c_void),
+            .enumerate()
+            .flat_map(|(buffer_idx, maybe_buffer)| match maybe_buffer {
+                Some(b) => {
+                    match (data.data_type(), buffer_idx) {
+                        (
+                            DataType::Utf8
+                            | DataType::LargeUtf8
+                            | DataType::Binary
+                            | DataType::LargeBinary,
+                            1,
+                        ) => {
+                            // For offset buffer, take original pointer without offset.
+                            // Buffer offset should be handled by `FFI_ArrowArray` offset field.
+                            Some(b.data_ptr().as_ptr() as *const c_void)
+                        }
+                        // For other buffers, note that `raw_data` takes into account the buffer's offset
+                        _ => Some(b.as_ptr() as *const c_void),
+                    }
+                }
                 // This is for null buffer. We only put a null pointer for
                 // null buffer if by spec it can contain null mask.
                 None if data_layout.can_contain_null_mask => Some(std::ptr::null()),
@@ -186,7 +233,7 @@ impl FFI_ArrowArray {
         Self {
             length: data.len() as i64,
             null_count: null_count as i64,
-            offset: data.offset() as i64,
+            offset: offset as i64,
             n_buffers,
             n_children,
             buffers: private_data.buffers_ptr.as_mut_ptr(),
diff --git a/arrow-data/src/transform/mod.rs b/arrow-data/src/transform/mod.rs
index b0d9475af..1e43bf488 100644
--- a/arrow-data/src/transform/mod.rs
+++ b/arrow-data/src/transform/mod.rs
@@ -385,7 +385,10 @@ impl<'a> MutableArrayData<'a> {
                 array_capacity = *capacity;
                 new_buffers(data_type, *capacity)
             }
-            (DataType::List(_) | DataType::LargeList(_), Capacities::List(capacity, _)) => {
+            (
+                DataType::List(_) | DataType::LargeList(_) | DataType::FixedSizeList(_, _),
+                Capacities::List(capacity, _),
+            ) => {
                 array_capacity = *capacity;
                 new_buffers(data_type, *capacity)
             }
@@ -501,12 +504,23 @@ impl<'a> MutableArrayData<'a> {
                     MutableArrayData::new(value_child, use_nulls, array_capacity),
                 ]
             }
-            DataType::FixedSizeList(_, _) => {
+            DataType::FixedSizeList(_, size) => {
                 let children = arrays
                     .iter()
                     .map(|array| &array.child_data()[0])
                     .collect::<Vec<_>>();
-                vec![MutableArrayData::new(children, use_nulls, array_capacity)]
+                let capacities =
+                    if let Capacities::List(capacity, ref child_capacities) = capacities {
+                        child_capacities
+                            .clone()
+                            .map(|c| *c)
+                            .unwrap_or(Capacities::Array(capacity * *size as usize))
+                    } else {
+                        Capacities::Array(array_capacity * *size as usize)
+                    };
+                vec![MutableArrayData::with_capacities(
+                    children, use_nulls, capacities,
+                )]
             }
             DataType::Union(fields, _) => (0..fields.len())
                 .map(|i| {
diff --git a/arrow-flight/gen/Cargo.toml b/arrow-flight/gen/Cargo.toml
index 8c4587cec..7264a527c 100644
--- a/arrow-flight/gen/Cargo.toml
+++ b/arrow-flight/gen/Cargo.toml
@@ -32,6 +32,6 @@ publish = false
 [dependencies]
 # 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.85", default-features = false }
+proc-macro2 = { version = "=1.0.86", default-features = false }
 prost-build = { version = "=0.12.6", default-features = false }
 tonic-build = { version = "=0.11.0", default-features = false, features = ["transport", "prost"] }
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index 3c203a7f3..3423d06b6 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -31,7 +31,7 @@ use std::io::{BufReader, Read, Seek, SeekFrom};
 use std::sync::Arc;
 
 use arrow_array::*;
-use arrow_buffer::{ArrowNativeType, Buffer, MutableBuffer, ScalarBuffer};
+use arrow_buffer::{ArrowNativeType, BooleanBuffer, Buffer, MutableBuffer, ScalarBuffer};
 use arrow_data::ArrayData;
 use arrow_schema::*;
 
@@ -152,7 +152,15 @@ fn create_array(
                 struct_arrays.push((struct_field.clone(), child));
             }
             let null_count = struct_node.null_count() as usize;
-            let struct_array = if null_count > 0 {
+            let struct_array = if struct_arrays.is_empty() {
+                // `StructArray::from` can't infer the correct row count
+                // if we have zero fields
+                let len = struct_node.length() as usize;
+                StructArray::new_empty_fields(
+                    len,
+                    (null_count > 0).then(|| BooleanBuffer::new(null_buffer, 0, len).into()),
+                )
+            } else if null_count > 0 {
                 // create struct array from fields, arrays and null data
                 StructArray::from((struct_arrays, null_buffer))
             } else {
@@ -1361,6 +1369,7 @@ mod tests {
     use crate::root_as_message;
     use arrow_array::builder::{PrimitiveRunBuilder, UnionBuilder};
     use arrow_array::types::*;
+    use arrow_buffer::NullBuffer;
     use arrow_data::ArrayDataBuilder;
 
     fn create_test_projection_schema() -> Schema {
@@ -1699,6 +1708,18 @@ mod tests {
         check_union_with_builder(UnionBuilder::new_sparse());
     }
 
+    #[test]
+    fn test_roundtrip_struct_empty_fields() {
+        let nulls = NullBuffer::from(&[true, true, false][..]);
+        let rb = RecordBatch::try_from_iter([(
+            "",
+            Arc::new(StructArray::new_empty_fields(nulls.len(), Some(nulls))) as _,
+        )])
+        .unwrap();
+        let rb2 = roundtrip_ipc(&rb);
+        assert_eq!(rb, rb2);
+    }
+
     #[test]
     fn test_roundtrip_stream_run_array_sliced() {
         let run_array_1: Int32RunArray = vec!["a", "a", "a", "b", "b", "c", "c", "c"]
diff --git a/arrow-select/src/concat.rs b/arrow-select/src/concat.rs
index f98e85475..5732d721b 100644
--- a/arrow-select/src/concat.rs
+++ b/arrow-select/src/concat.rs
@@ -54,6 +54,34 @@ fn binary_capacity<T: ByteArrayType>(arrays: &[&dyn Array]) -> Capacities {
     Capacities::Binary(item_capacity, Some(bytes_capacity))
 }
 
+fn fixed_size_list_capacity(arrays: &[&dyn Array], data_type: &DataType) -> Capacities {
+    if let DataType::FixedSizeList(f, _) = data_type {
+        let item_capacity = arrays.iter().map(|a| a.len()).sum();
+        let child_data_type = f.data_type();
+        match child_data_type {
+            // These types should match the types that `get_capacity`
+            // has special handling for.
+            DataType::Utf8
+            | DataType::LargeUtf8
+            | DataType::Binary
+            | DataType::LargeBinary
+            | DataType::FixedSizeList(_, _) => {
+                let values: Vec<&dyn arrow_array::Array> = arrays
+                    .iter()
+                    .map(|a| a.as_fixed_size_list().values().as_ref())
+                    .collect();
+                Capacities::List(
+                    item_capacity,
+                    Some(Box::new(get_capacity(&values, child_data_type))),
+                )
+            }
+            _ => Capacities::Array(item_capacity),
+        }
+    } else {
+        unreachable!("illegal data type for fixed size list")
+    }
+}
+
 fn concat_dictionaries<K: ArrowDictionaryKeyType>(
     arrays: &[&dyn Array],
 ) -> Result<ArrayRef, ArrowError> {
@@ -107,6 +135,17 @@ macro_rules! dict_helper {
     };
 }
 
+fn get_capacity(arrays: &[&dyn Array], data_type: &DataType) -> Capacities {
+    match data_type {
+        DataType::Utf8 => binary_capacity::<Utf8Type>(arrays),
+        DataType::LargeUtf8 => binary_capacity::<LargeUtf8Type>(arrays),
+        DataType::Binary => binary_capacity::<BinaryType>(arrays),
+        DataType::LargeBinary => binary_capacity::<LargeBinaryType>(arrays),
+        DataType::FixedSizeList(_, _) => fixed_size_list_capacity(arrays, data_type),
+        _ => Capacities::Array(arrays.iter().map(|a| a.len()).sum()),
+    }
+}
+
 /// Concatenate multiple [Array] of the same type into a single [ArrayRef].
 pub fn concat(arrays: &[&dyn Array]) -> Result<ArrayRef, ArrowError> {
     if arrays.is_empty() {
@@ -124,20 +163,15 @@ pub fn concat(arrays: &[&dyn Array]) -> Result<ArrayRef, ArrowError> {
             "It is not possible to concatenate arrays of different data types.".to_string(),
         ));
     }
-
-    let capacity = match d {
-        DataType::Utf8 => binary_capacity::<Utf8Type>(arrays),
-        DataType::LargeUtf8 => binary_capacity::<LargeUtf8Type>(arrays),
-        DataType::Binary => binary_capacity::<BinaryType>(arrays),
-        DataType::LargeBinary => binary_capacity::<LargeBinaryType>(arrays),
-        DataType::Dictionary(k, _) => downcast_integer! {
+    if let DataType::Dictionary(k, _) = d {
+        downcast_integer! {
             k.as_ref() => (dict_helper, arrays),
             _ => unreachable!("illegal dictionary key type {k}")
-        },
-        _ => Capacities::Array(arrays.iter().map(|a| a.len()).sum()),
-    };
-
-    concat_fallback(arrays, capacity)
+        };
+    } else {
+        let capacity = get_capacity(arrays, d);
+        concat_fallback(arrays, capacity)
+    }
 }
 
 /// Concatenates arrays using MutableArrayData
@@ -373,6 +407,37 @@ mod tests {
         assert_eq!(array_result.as_ref(), &array_expected as &dyn Array);
     }
 
+    #[test]
+    fn test_concat_primitive_fixed_size_list_arrays() {
+        let list1 = vec![
+            Some(vec![Some(-1), None]),
+            None,
+            Some(vec![Some(10), Some(20)]),
+        ];
+        let list1_array =
+            FixedSizeListArray::from_iter_primitive::<Int64Type, _, _>(list1.clone(), 2);
+
+        let list2 = vec![
+            None,
+            Some(vec![Some(100), None]),
+            Some(vec![Some(102), Some(103)]),
+        ];
+        let list2_array =
+            FixedSizeListArray::from_iter_primitive::<Int64Type, _, _>(list2.clone(), 2);
+
+        let list3 = vec![Some(vec![Some(1000), Some(1001)])];
+        let list3_array =
+            FixedSizeListArray::from_iter_primitive::<Int64Type, _, _>(list3.clone(), 2);
+
+        let array_result = concat(&[&list1_array, &list2_array, &list3_array]).unwrap();
+
+        let expected = list1.into_iter().chain(list2).chain(list3);
+        let array_expected =
+            FixedSizeListArray::from_iter_primitive::<Int64Type, _, _>(expected, 2);
+
+        assert_eq!(array_result.as_ref(), &array_expected as &dyn Array);
+    }
+
     #[test]
     fn test_concat_struct_arrays() {
         let field = Arc::new(Field::new("field", DataType::Int64, true));
diff --git a/arrow/benches/comparison_kernels.rs b/arrow/benches/comparison_kernels.rs
index f330e1386..1e081d141 100644
--- a/arrow/benches/comparison_kernels.rs
+++ b/arrow/benches/comparison_kernels.rs
@@ -17,6 +17,7 @@
 
 #[macro_use]
 extern crate criterion;
+use arrow::util::test_util::seedable_rng;
 use criterion::Criterion;
 
 extern crate arrow;
@@ -27,6 +28,8 @@ use arrow::{array::*, datatypes::Float32Type, datatypes::Int32Type};
 use arrow_buffer::IntervalMonthDayNano;
 use arrow_string::like::*;
 use arrow_string::regexp::regexp_is_match_utf8_scalar;
+use rand::rngs::StdRng;
+use rand::Rng;
 
 const SIZE: usize = 65536;
 
@@ -55,6 +58,14 @@ fn bench_regexp_is_match_utf8_scalar(arr_a: &StringArray, value_b: &str) {
     .unwrap();
 }
 
+fn make_string_array(size: usize, rng: &mut StdRng) -> impl Iterator<Item = Option<String>> + '_ {
+    (0..size).map(|_| {
+        let len = rng.gen_range(0..64);
+        let bytes = (0..len).map(|_| rng.gen_range(0..128)).collect();
+        Some(String::from_utf8(bytes).unwrap())
+    })
+}
+
 fn add_benchmark(c: &mut Criterion) {
     let arr_a = create_primitive_array_with_seed::<Float32Type>(SIZE, 0.0, 42);
     let arr_b = create_primitive_array_with_seed::<Float32Type>(SIZE, 0.0, 43);
@@ -63,6 +74,7 @@ fn add_benchmark(c: &mut Criterion) {
     let arr_month_day_nano_b = create_month_day_nano_array_with_seed(SIZE, 0.0, 43);
 
     let arr_string = create_string_array::<i32>(SIZE, 0.0);
+
     let scalar = Float32Array::from(vec![1.0]);
 
     c.bench_function("eq Float32", |b| b.iter(|| eq(&arr_a, &arr_b)));
@@ -138,6 +150,45 @@ fn add_benchmark(c: &mut Criterion) {
         b.iter(|| eq(&arr_month_day_nano_b, &scalar).unwrap())
     });
 
+    let mut rng = seedable_rng();
+    let mut array_gen = make_string_array(1024 * 1024 * 8, &mut rng);
+    let string_left = StringArray::from_iter(array_gen);
+    let string_view_left = StringViewArray::from_iter(string_left.iter());
+
+    // reference to the same rng to make sure we generate **different** array data,
+    // ow. the left and right will be identical
+    array_gen = make_string_array(1024 * 1024 * 8, &mut rng);
+    let string_right = StringArray::from_iter(array_gen);
+    let string_view_right = StringViewArray::from_iter(string_right.iter());
+
+    c.bench_function("eq scalar StringArray", |b| {
+        b.iter(|| {
+            eq(
+                &Scalar::new(StringArray::from_iter_values(["xxxx"])),
+                &string_left,
+            )
+            .unwrap()
+        })
+    });
+
+    c.bench_function("eq scalar StringViewArray", |b| {
+        b.iter(|| {
+            eq(
+                &Scalar::new(StringViewArray::from_iter_values(["xxxx"])),
+                &string_view_left,
+            )
+            .unwrap()
+        })
+    });
+
+    c.bench_function("eq StringArray StringArray", |b| {
+        b.iter(|| eq(&string_left, &string_right).unwrap())
+    });
+
+    c.bench_function("eq StringViewArray StringViewArray", |b| {
+        b.iter(|| eq(&string_view_left, &string_view_right).unwrap())
+    });
+
     c.bench_function("like_utf8 scalar equals", |b| {
         b.iter(|| bench_like_utf8_scalar(&arr_string, "xxxx"))
     });
diff --git a/arrow/benches/concatenate_kernel.rs b/arrow/benches/concatenate_kernel.rs
index 2f5b65439..0c553f8b3 100644
--- a/arrow/benches/concatenate_kernel.rs
+++ b/arrow/benches/concatenate_kernel.rs
@@ -17,6 +17,8 @@
 
 #[macro_use]
 extern crate criterion;
+use std::sync::Arc;
+
 use criterion::Criterion;
 
 extern crate arrow;
@@ -82,6 +84,24 @@ fn add_benchmark(c: &mut Criterion) {
     c.bench_function("concat str nulls 1024", |b| {
         b.iter(|| bench_concat(&v1, &v2))
     });
+
+    let v1 = FixedSizeListArray::try_new(
+        Arc::new(Field::new("item", DataType::Int32, true)),
+        1024,
+        Arc::new(create_primitive_array::<Int32Type>(1024 * 1024, 0.0)),
+        None,
+    )
+    .unwrap();
+    let v2 = FixedSizeListArray::try_new(
+        Arc::new(Field::new("item", DataType::Int32, true)),
+        1024,
+        Arc::new(create_primitive_array::<Int32Type>(1024 * 1024, 0.0)),
+        None,
+    )
+    .unwrap();
+    c.bench_function("concat fixed size lists", |b| {
+        b.iter(|| bench_concat(&v1, &v2))
+    });
 }
 
 criterion_group!(benches, add_benchmark);
diff --git a/object_store/src/upload.rs b/object_store/src/upload.rs
index e5f683a03..dc499e2e0 100644
--- a/object_store/src/upload.rs
+++ b/object_store/src/upload.rs
@@ -92,6 +92,21 @@ pub trait MultipartUpload: Send + std::fmt::Debug {
     async fn abort(&mut self) -> Result<()>;
 }
 
+#[async_trait]
+impl<W: MultipartUpload + ?Sized> MultipartUpload for Box<W> {
+    fn put_part(&mut self, data: PutPayload) -> UploadPart {
+        (**self).put_part(data)
+    }
+
+    async fn complete(&mut self) -> Result<PutResult> {
+        (**self).complete().await
+    }
+
+    async fn abort(&mut self) -> Result<()> {
+        (**self).abort().await
+    }
+}
+
 /// A synchronous write API for uploading data in parallel in fixed size chunks
 ///
 /// Uses multiple tokio tasks in a [`JoinSet`] to multiplex upload tasks in parallel
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 775ac825a..7df978016 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -86,10 +86,10 @@ object_store = { version = "0.10.0", default-features = false, features = ["azur
 # TODO: temporary to fix parquet wasm build
 # upstream issue: https://github.com/gyscos/zstd-rs/issues/269
 [target.'cfg(target_family = "wasm")'.dependencies]
-zstd-sys = { version = ">=2.0.0, <2.0.10", optional = true, default-features = false }
+zstd-sys = { version = ">=2.0.0, <2.0.12", optional = true, default-features = false }
 
 [target.'cfg(target_family = "wasm")'.dev-dependencies]
-zstd-sys = { version = ">=2.0.0, <2.0.10", default-features = false }
+zstd-sys = { version = ">=2.0.0, <2.0.12", default-features = false }
 
 [package.metadata.docs.rs]
 all-features = true


Test patch:
diff --git a/arrow/tests/array_transform.rs b/arrow/tests/array_transform.rs
index 42e4da7c4..08f23c200 100644
--- a/arrow/tests/array_transform.rs
+++ b/arrow/tests/array_transform.rs
@@ -17,9 +17,9 @@
 
 use arrow::array::{
     Array, ArrayRef, BooleanArray, Decimal128Array, DictionaryArray, FixedSizeBinaryArray,
-    Int16Array, Int32Array, Int64Array, Int64Builder, ListArray, ListBuilder, MapBuilder,
-    NullArray, StringArray, StringBuilder, StringDictionaryBuilder, StructArray, UInt8Array,
-    UnionArray,
+    FixedSizeListBuilder, Int16Array, Int32Array, Int64Array, Int64Builder, ListArray, ListBuilder,
+    MapBuilder, NullArray, StringArray, StringBuilder, StringDictionaryBuilder, StructArray,
+    UInt16Array, UInt16Builder, UInt8Array, UnionArray,
 };
 use arrow::datatypes::Int16Type;
 use arrow_array::StringViewArray;
@@ -1074,43 +1074,42 @@ fn test_mixed_types() {
     MutableArrayData::new(vec![&a, &b], false, 4);
 }
 
-/*
-// this is an old test used on a meanwhile removed dead code
-// that is still useful when `MutableArrayData` supports fixed-size lists.
 #[test]
-fn test_fixed_size_list_append() -> Result<()> {
-    let int_builder = UInt16Builder::new(64);
+fn test_fixed_size_list_append() {
+    let int_builder = UInt16Builder::with_capacity(64);
     let mut builder = FixedSizeListBuilder::<UInt16Builder>::new(int_builder, 2);
-    builder.values().append_slice(&[1, 2])?;
-    builder.append(true)?;
-    builder.values().append_slice(&[3, 4])?;
-    builder.append(false)?;
-    builder.values().append_slice(&[5, 6])?;
-    builder.append(true)?;
-
-    let a_builder = UInt16Builder::new(64);
+    builder.values().append_slice(&[1, 2]);
+    builder.append(true);
+    builder.values().append_slice(&[3, 4]);
+    builder.append(false);
+    builder.values().append_slice(&[5, 6]);
+    builder.append(true);
+    let a = builder.finish().into_data();
+
+    let a_builder = UInt16Builder::with_capacity(64);
     let mut a_builder = FixedSizeListBuilder::<UInt16Builder>::new(a_builder, 2);
-    a_builder.values().append_slice(&[7, 8])?;
-    a_builder.append(true)?;
-    a_builder.values().append_slice(&[9, 10])?;
-    a_builder.append(true)?;
-    a_builder.values().append_slice(&[11, 12])?;
-    a_builder.append(false)?;
-    a_builder.values().append_slice(&[13, 14])?;
-    a_builder.append(true)?;
-    a_builder.values().append_null()?;
-    a_builder.values().append_null()?;
-    a_builder.append(true)?;
-    let a = a_builder.finish();
+    a_builder.values().append_slice(&[7, 8]);
+    a_builder.append(true);
+    a_builder.values().append_slice(&[9, 10]);
+    a_builder.append(true);
+    a_builder.values().append_slice(&[11, 12]);
+    a_builder.append(false);
+    a_builder.values().append_slice(&[13, 14]);
+    a_builder.append(true);
+    a_builder.values().append_null();
+    a_builder.values().append_null();
+    a_builder.append(true);
+    let b = a_builder.finish().into_data();
+
+    let mut mutable = MutableArrayData::new(vec![&a, &b], false, 10);
+    mutable.extend(0, 0, a.len());
+    mutable.extend(1, 0, b.len());
 
     // append array
-    builder.append_data(&[
-        a.data(),
-        a.slice(1, 3).data(),
-        a.slice(2, 1).data(),
-        a.slice(5, 0).data(),
-    ])?;
-    let finished = builder.finish();
+    mutable.extend(1, 1, 4);
+    mutable.extend(1, 2, 3);
+
+    let finished = mutable.freeze();
 
     let expected_int_array = UInt16Array::from(vec![
         Some(1),
@@ -1141,23 +1140,14 @@ fn test_fixed_size_list_append() -> Result<()> {
         Some(11),
         Some(12),
     ]);
-    let expected_list_data = ArrayData::new(
-        DataType::FixedSizeList(
-            Arc::new(Field::new("item", DataType::UInt16, true)),
-            2,
-        ),
+    let expected_fixed_size_list_data = ArrayData::try_new(
+        DataType::FixedSizeList(Arc::new(Field::new("item", DataType::UInt16, true)), 2),
         12,
-        None,
-        None,
+        Some(Buffer::from(&[0b11011101, 0b101])),
         0,
         vec![],
-        vec![expected_int_array.data()],
-    );
-    let expected_list =
-        FixedSizeListArray::from(Arc::new(expected_list_data) as ArrayData);
-    assert_eq!(&expected_list.values(), &finished.values());
-    assert_eq!(expected_list.len(), finished.len());
-
-    Ok(())
+        vec![expected_int_array.to_data()],
+    )
+    .unwrap();
+    assert_eq!(finished, expected_fixed_size_list_data);
 }
-*/