Repository: apache/arrow-rs
Pull request #5875 (2024-06-12 15:58:02)

Document parquet ArrowWriter type limitations
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes https://github.com/apache/arrow-rs/issues/5849
Closes https://github.com/apache/arrow-rs/pull/5868

# Rationale for this change
It was not clear that the parquet writer is not able to write IntervalMonthDayNano data rather than it was not yet implemented

# What changes are included in this PR?
Add documentiation explaining type support to `ArrowWriter`

<!--
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?

Docs only



alamb: cc @xinlifoobar 

Patch:
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index b00d2c88e..a8dbeded9 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -62,3 +62,7 @@ criterion = { version = "0.5", default-features = false }
 [[bench]]
 name = "occupancy"
 harness = false
+
+[[bench]]
+name = "gc_view_types"
+harness = false
diff --git a/arrow-array/benches/gc_view_types.rs b/arrow-array/benches/gc_view_types.rs
new file mode 100644
index 000000000..4b74a8f60
--- /dev/null
+++ b/arrow-array/benches/gc_view_types.rs
@@ -0,0 +1,48 @@
+// 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 arrow_array::StringViewArray;
+use criterion::*;
+
+fn gen_view_array(size: usize) -> StringViewArray {
+    StringViewArray::from_iter((0..size).map(|v| match v % 3 {
+        0 => Some("small"),
+        1 => Some("larger than 12 bytes array"),
+        2 => None,
+        _ => unreachable!("unreachable"),
+    }))
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let array = gen_view_array(100_000);
+
+    c.bench_function("gc view types all", |b| {
+        b.iter(|| {
+            black_box(array.gc());
+        });
+    });
+
+    let sliced = array.slice(0, 100_000 / 2);
+    c.bench_function("gc view types slice half", |b| {
+        b.iter(|| {
+            black_box(sliced.gc());
+        });
+    });
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index 79f2d4758..88033b83c 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -265,6 +265,56 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
             phantom: Default::default(),
         }
     }
+
+    /// Returns a "compacted" version of this array
+    ///
+    /// The original array will *not* be modified
+    ///
+    /// # Garbage Collection
+    ///
+    /// Before GC:
+    /// ```text
+    ///                                        ┌──────┐                 
+    ///                                        │......│                 
+    ///                                        │......│                 
+    /// ┌────────────────────┐       ┌ ─ ─ ─ ▶ │Data1 │   Large buffer  
+    /// │       View 1       │─ ─ ─ ─          │......│  with data that
+    /// ├────────────────────┤                 │......│ is not referred
+    /// │       View 2       │─ ─ ─ ─ ─ ─ ─ ─▶ │Data2 │ to by View 1 or
+    /// └────────────────────┘                 │......│      View 2     
+    ///                                        │......│                 
+    ///    2 views, refer to                   │......│                 
+    ///   small portions of a                  └──────┘                 
+    ///      large buffer                                               
+    /// ```
+    ///                                                                
+    /// After GC:
+    ///
+    /// ```text
+    /// ┌────────────────────┐                 ┌─────┐    After gc, only
+    /// │       View 1       │─ ─ ─ ─ ─ ─ ─ ─▶ │Data1│     data that is  
+    /// ├────────────────────┤       ┌ ─ ─ ─ ▶ │Data2│    pointed to by  
+    /// │       View 2       │─ ─ ─ ─          └─────┘     the views is  
+    /// └────────────────────┘                                 left      
+    ///                                                                  
+    ///                                                                  
+    ///         2 views                                                  
+    /// ```
+    /// This method will compact the data buffers by recreating the view array and only include the data
+    /// that is pointed to by the views.
+    ///
+    /// Note that it will copy the array regardless of whether the original array is compact.
+    /// Use with caution as this can be an expensive operation, only use it when you are sure that the view
+    /// array is significantly smaller than when it is originally created, e.g., after filtering or slicing.
+    pub fn gc(&self) -> Self {
+        let mut builder = GenericByteViewBuilder::<T>::with_capacity(self.len());
+
+        for v in self.iter() {
+            builder.append_option(v);
+        }
+
+        builder.finish()
+    }
 }
 
 impl<T: ByteViewType + ?Sized> Debug for GenericByteViewArray<T> {
@@ -645,4 +695,39 @@ mod tests {
 
         StringViewArray::new(views, buffers, None);
     }
+
+    #[test]
+    fn test_gc() {
+        let test_data = [
+            Some("longer than 12 bytes"),
+            Some("short"),
+            Some("t"),
+            Some("longer than 12 bytes"),
+            None,
+            Some("short"),
+        ];
+
+        let array = {
+            let mut builder = StringViewBuilder::new().with_block_size(8); // create multiple buffers
+            test_data.into_iter().for_each(|v| builder.append_option(v));
+            builder.finish()
+        };
+        assert!(array.buffers.len() > 1);
+
+        fn check_gc(to_test: &StringViewArray) {
+            let gc = to_test.gc();
+            assert_ne!(to_test.data_buffers().len(), gc.data_buffers().len());
+
+            to_test.iter().zip(gc.iter()).for_each(|(a, b)| {
+                assert_eq!(a, b);
+            });
+            assert_eq!(to_test.len(), gc.len());
+        }
+
+        check_gc(&array);
+        check_gc(&array.slice(1, 3));
+        check_gc(&array.slice(2, 1));
+        check_gc(&array.slice(2, 2));
+        check_gc(&array.slice(3, 1));
+    }
 }
diff --git a/arrow-array/src/builder/generic_bytes_view_builder.rs b/arrow-array/src/builder/generic_bytes_view_builder.rs
index e7f13a682..6ec34bf5a 100644
--- a/arrow-array/src/builder/generic_bytes_view_builder.rs
+++ b/arrow-array/src/builder/generic_bytes_view_builder.rs
@@ -116,6 +116,36 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
         offset as u32
     }
 
+    /// Append a view of the given `block`, `offset` and `length`
+    ///
+    /// # Safety
+    /// (1) The block must have been added using [`Self::append_block`]
+    /// (2) The range `offset..offset+length` must be within the bounds of the block
+    /// (3) The data in the block must be valid of type `T`
+    pub unsafe fn append_view_unchecked(&mut self, block: u32, offset: u32, len: u32) {
+        let b = self.completed.get_unchecked(block as usize);
+        let start = offset as usize;
+        let end = start.saturating_add(len as usize);
+        let b = b.get_unchecked(start..end);
+
+        if len <= 12 {
+            let mut view_buffer = [0; 16];
+            view_buffer[0..4].copy_from_slice(&len.to_le_bytes());
+            view_buffer[4..4 + b.len()].copy_from_slice(b);
+            self.views_builder.append(u128::from_le_bytes(view_buffer));
+        } else {
+            let view = ByteView {
+                length: len,
+                prefix: u32::from_le_bytes(b[0..4].try_into().unwrap()),
+                buffer_index: block,
+                offset,
+            };
+            self.views_builder.append(view.into());
+        }
+
+        self.null_buffer_builder.append_non_null();
+    }
+
     /// Try to append a view of the given `block`, `offset` and `length`
     ///
     /// See [`Self::append_block`]
@@ -139,22 +169,9 @@ impl<T: ByteViewType + ?Sized> GenericByteViewBuilder<T> {
             ));
         }
 
-        if len <= 12 {
-            let mut view_buffer = [0; 16];
-            view_buffer[0..4].copy_from_slice(&len.to_le_bytes());
-            view_buffer[4..4 + b.len()].copy_from_slice(b);
-            self.views_builder.append(u128::from_le_bytes(view_buffer));
-        } else {
-            let view = ByteView {
-                length: len,
-                prefix: u32::from_le_bytes(b[0..4].try_into().unwrap()),
-                buffer_index: block,
-                offset,
-            };
-            self.views_builder.append(view.into());
+        unsafe {
+            self.append_view_unchecked(block, offset, len);
         }
-
-        self.null_buffer_builder.append_non_null();
         Ok(())
     }
 
diff --git a/arrow-array/src/run_iterator.rs b/arrow-array/src/run_iterator.rs
index 7a98fccb7..2922bf04d 100644
--- a/arrow-array/src/run_iterator.rs
+++ b/arrow-array/src/run_iterator.rs
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Idiomatic iterator for [`RunArray`](crate::Array)
+//! Idiomatic iterator for [`RunArray`](crate::RunArray)
 
 use crate::{array::ArrayAccessor, types::RunEndIndexType, Array, TypedRunArray};
 use arrow_buffer::ArrowNativeType;
diff --git a/arrow-cast/src/cast/dictionary.rs b/arrow-cast/src/cast/dictionary.rs
index 244e101f1..ee2021d15 100644
--- a/arrow-cast/src/cast/dictionary.rs
+++ b/arrow-cast/src/cast/dictionary.rs
@@ -85,10 +85,69 @@ pub(crate) fn dictionary_cast<K: ArrowDictionaryKeyType>(
 
             Ok(new_array)
         }
+        Utf8View => {
+            // `unpack_dictionary` can handle Utf8View/BinaryView types, but incurs unnecessary data copy of the value buffer.
+            // we handle it here to avoid the copy.
+            let dict_array = array
+                .as_dictionary::<K>()
+                .downcast_dict::<StringArray>()
+                .unwrap();
+
+            let string_view = view_from_dict_values::<K, StringViewType, GenericStringType<i32>>(
+                dict_array.values(),
+                dict_array.keys(),
+            );
+            Ok(Arc::new(string_view))
+        }
+        BinaryView => {
+            // `unpack_dictionary` can handle Utf8View/BinaryView types, but incurs unnecessary data copy of the value buffer.
+            // we handle it here to avoid the copy.
+            let dict_array = array
+                .as_dictionary::<K>()
+                .downcast_dict::<BinaryArray>()
+                .unwrap();
+
+            let binary_view = view_from_dict_values::<K, BinaryViewType, BinaryType>(
+                dict_array.values(),
+                dict_array.keys(),
+            );
+            Ok(Arc::new(binary_view))
+        }
         _ => unpack_dictionary::<K>(array, to_type, cast_options),
     }
 }
 
+fn view_from_dict_values<K: ArrowDictionaryKeyType, T: ByteViewType, V: ByteArrayType>(
+    array: &GenericByteArray<V>,
+    keys: &PrimitiveArray<K>,
+) -> GenericByteViewArray<T> {
+    let value_buffer = array.values();
+    let value_offsets = array.value_offsets();
+    let mut builder = GenericByteViewBuilder::<T>::with_capacity(keys.len());
+    builder.append_block(value_buffer.clone());
+    for i in keys.iter() {
+        match i {
+            Some(v) => {
+                let idx = v.to_usize().unwrap();
+
+                // Safety
+                // (1) The index is within bounds as they are offsets
+                // (2) The append_view is safe
+                unsafe {
+                    let offset = value_offsets.get_unchecked(idx).as_usize();
+                    let end = value_offsets.get_unchecked(idx + 1).as_usize();
+                    let length = end - offset;
+                    builder.append_view_unchecked(0, offset as u32, length as u32)
+                }
+            }
+            None => {
+                builder.append_null();
+            }
+        }
+    }
+    builder.finish()
+}
+
 // Unpack a dictionary where the keys are of type <K> into a flattened array of type to_type
 pub(crate) fn unpack_dictionary<K>(
     array: &dyn Array,
@@ -129,10 +188,34 @@ pub(crate) fn cast_to_dictionary<K: ArrowDictionaryKeyType>(
         Decimal256(_, _) => {
             pack_numeric_to_dictionary::<K, Decimal256Type>(array, dict_value_type, cast_options)
         }
-        Utf8 => pack_byte_to_dictionary::<K, GenericStringType<i32>>(array, cast_options),
-        LargeUtf8 => pack_byte_to_dictionary::<K, GenericStringType<i64>>(array, cast_options),
-        Binary => pack_byte_to_dictionary::<K, GenericBinaryType<i32>>(array, cast_options),
-        LargeBinary => pack_byte_to_dictionary::<K, GenericBinaryType<i64>>(array, cast_options),
+        Utf8 => {
+            // If the input is a view type, we can avoid casting (thus copying) the data
+            if array.data_type() == &DataType::Utf8View {
+                return string_view_to_dictionary::<K, i32>(array);
+            }
+            pack_byte_to_dictionary::<K, GenericStringType<i32>>(array, cast_options)
+        }
+        LargeUtf8 => {
+            // If the input is a view type, we can avoid casting (thus copying) the data
+            if array.data_type() == &DataType::Utf8View {
+                return string_view_to_dictionary::<K, i64>(array);
+            }
+            pack_byte_to_dictionary::<K, GenericStringType<i64>>(array, cast_options)
+        }
+        Binary => {
+            // If the input is a view type, we can avoid casting (thus copying) the data
+            if array.data_type() == &DataType::BinaryView {
+                return binary_view_to_dictionary::<K, i32>(array);
+            }
+            pack_byte_to_dictionary::<K, GenericBinaryType<i32>>(array, cast_options)
+        }
+        LargeBinary => {
+            // If the input is a view type, we can avoid casting (thus copying) the data
+            if array.data_type() == &DataType::BinaryView {
+                return binary_view_to_dictionary::<K, i64>(array);
+            }
+            pack_byte_to_dictionary::<K, GenericBinaryType<i64>>(array, cast_options)
+        }
         _ => Err(ArrowError::CastError(format!(
             "Unsupported output type for dictionary packing: {dict_value_type:?}"
         ))),
@@ -167,6 +250,58 @@ where
     Ok(Arc::new(b.finish()))
 }
 
+pub(crate) fn string_view_to_dictionary<K, O: OffsetSizeTrait>(
+    array: &dyn Array,
+) -> Result<ArrayRef, ArrowError>
+where
+    K: ArrowDictionaryKeyType,
+{
+    let mut b = GenericByteDictionaryBuilder::<K, GenericStringType<O>>::with_capacity(
+        array.len(),
+        1024,
+        1024,
+    );
+    let string_view = array.as_any().downcast_ref::<StringViewArray>().unwrap();
+    for v in string_view.iter() {
+        match v {
+            Some(v) => {
+                b.append(v)?;
+            }
+            None => {
+                b.append_null();
+            }
+        }
+    }
+
+    Ok(Arc::new(b.finish()))
+}
+
+pub(crate) fn binary_view_to_dictionary<K, O: OffsetSizeTrait>(
+    array: &dyn Array,
+) -> Result<ArrayRef, ArrowError>
+where
+    K: ArrowDictionaryKeyType,
+{
+    let mut b = GenericByteDictionaryBuilder::<K, GenericBinaryType<O>>::with_capacity(
+        array.len(),
+        1024,
+        1024,
+    );
+    let binary_view = array.as_any().downcast_ref::<BinaryViewArray>().unwrap();
+    for v in binary_view.iter() {
+        match v {
+            Some(v) => {
+                b.append(v)?;
+            }
+            None => {
+                b.append_null();
+            }
+        }
+    }
+
+    Ok(Arc::new(b.finish()))
+}
+
 // Packs the data as a GenericByteDictionaryBuilder, if possible, with the
 // key types of K
 pub(crate) fn pack_byte_to_dictionary<K, T>(
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index ae2240bd9..55f2ed728 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -46,7 +46,7 @@ use crate::cast::dictionary::*;
 use crate::cast::list::*;
 use crate::cast::string::*;
 
-use arrow_buffer::{IntervalMonthDayNano, ScalarBuffer};
+use arrow_buffer::IntervalMonthDayNano;
 use arrow_data::ByteView;
 use chrono::{NaiveTime, Offset, TimeZone, Utc};
 use std::cmp::Ordering;
@@ -2341,47 +2341,29 @@ where
     FROM::Offset: OffsetSizeTrait + ToPrimitive,
     V: ByteViewType,
 {
-    let data = array.to_data();
-    assert_eq!(data.data_type(), &FROM::DATA_TYPE);
-
+    let byte_array: &GenericByteArray<FROM> = array.as_bytes();
     let len = array.len();
-    let str_values_buf = data.buffers()[1].clone();
-    let offsets = data.buffers()[0].typed_data::<FROM::Offset>();
+    let str_values_buf = byte_array.values().clone();
+    let offsets = byte_array.offsets();
 
-    let mut views_builder = BufferBuilder::<u128>::new(len);
-    for w in offsets.windows(2) {
+    let mut views_builder = GenericByteViewBuilder::<V>::with_capacity(len);
+    let block = views_builder.append_block(str_values_buf);
+    for (i, w) in offsets.windows(2).enumerate() {
         let offset = w[0].to_u32().unwrap();
         let end = w[1].to_u32().unwrap();
-        let value_buf = &str_values_buf[offset as usize..end as usize];
         let length = end - offset;
 
-        if length <= 12 {
-            let mut view_buffer = [0; 16];
-            view_buffer[0..4].copy_from_slice(&length.to_le_bytes());
-            view_buffer[4..4 + value_buf.len()].copy_from_slice(value_buf);
-            views_builder.append(u128::from_le_bytes(view_buffer));
+        if byte_array.is_null(i) {
+            views_builder.append_null();
         } else {
-            let view = ByteView {
-                length,
-                prefix: u32::from_le_bytes(value_buf[0..4].try_into().unwrap()),
-                buffer_index: 0,
-                offset,
-            };
-            views_builder.append(view.into());
+            // Safety: the input was a valid array so it valid UTF8 (if string). And
+            // all offsets were valid and we created the views correctly
+            unsafe { views_builder.append_view_unchecked(block, offset, length) }
         }
     }
 
     assert_eq!(views_builder.len(), len);
-
-    // Safety: the input was a valid array so it valid UTF8 (if string). And
-    // all offsets were valid and we created the views correctly
-    Ok(Arc::new(unsafe {
-        GenericByteViewArray::<V>::new_unchecked(
-            ScalarBuffer::new(views_builder.finish(), 0, len),
-            vec![str_values_buf],
-            data.nulls().cloned(),
-        )
-    }))
+    Ok(Arc::new(views_builder.finish()))
 }
 
 /// Helper function to cast from one `ByteViewType` array to `ByteArrayType` array.
@@ -5203,19 +5185,19 @@ mod tests {
         _test_string_to_view::<i64>();
     }
 
+    const VIEW_TEST_DATA: [Option<&str>; 5] = [
+        Some("hello"),
+        Some("repeated"),
+        None,
+        Some("large payload over 12 bytes"),
+        Some("repeated"),
+    ];
+
     fn _test_string_to_view<O>()
     where
         O: OffsetSizeTrait,
     {
-        let data = vec![
-            Some("hello"),
-            Some("world"),
-            None,
-            Some("large payload over 12 bytes"),
-            Some("lulu"),
-        ];
-
-        let string_array = GenericStringArray::<O>::from(data.clone());
+        let string_array = GenericStringArray::<O>::from_iter(VIEW_TEST_DATA);
 
         assert!(can_cast_types(
             string_array.data_type(),
@@ -5225,7 +5207,7 @@ mod tests {
         let string_view_array = cast(&string_array, &DataType::Utf8View).unwrap();
         assert_eq!(string_view_array.data_type(), &DataType::Utf8View);
 
-        let expect_string_view_array = StringViewArray::from(data);
+        let expect_string_view_array = StringViewArray::from_iter(VIEW_TEST_DATA);
         assert_eq!(string_view_array.as_ref(), &expect_string_view_array);
     }
 
@@ -5239,15 +5221,7 @@ mod tests {
     where
         O: OffsetSizeTrait,
     {
-        let data: Vec<Option<&[u8]>> = vec![
-            Some(b"hello"),
-            Some(b"world"),
-            None,
-            Some(b"large payload over 12 bytes"),
-            Some(b"lulu"),
-        ];
-
-        let binary_array = GenericBinaryArray::<O>::from(data.clone());
+        let binary_array = GenericBinaryArray::<O>::from_iter(VIEW_TEST_DATA);
 
         assert!(can_cast_types(
             binary_array.data_type(),
@@ -5257,10 +5231,68 @@ mod tests {
         let binary_view_array = cast(&binary_array, &DataType::BinaryView).unwrap();
         assert_eq!(binary_view_array.data_type(), &DataType::BinaryView);
 
-        let expect_binary_view_array = BinaryViewArray::from(data);
+        let expect_binary_view_array = BinaryViewArray::from_iter(VIEW_TEST_DATA);
         assert_eq!(binary_view_array.as_ref(), &expect_binary_view_array);
     }
 
+    #[test]
+    fn test_dict_to_view() {
+        let values = StringArray::from_iter(VIEW_TEST_DATA);
+        let keys = Int8Array::from_iter([Some(1), Some(0), None, Some(3), None, Some(1), Some(4)]);
+        let string_dict_array =
+            DictionaryArray::<Int8Type>::try_new(keys, Arc::new(values)).unwrap();
+        let typed_dict = string_dict_array.downcast_dict::<StringArray>().unwrap();
+
+        let string_view_array = {
+            let mut builder = StringViewBuilder::new().with_block_size(8); // multiple buffers.
+            for v in typed_dict.into_iter() {
+                builder.append_option(v);
+            }
+            builder.finish()
+        };
+        let expected_string_array_type = string_view_array.data_type();
+        let casted_string_array = cast(&string_dict_array, expected_string_array_type).unwrap();
+        assert_eq!(casted_string_array.data_type(), expected_string_array_type);
+        assert_eq!(casted_string_array.as_ref(), &string_view_array);
+
+        let binary_buffer = cast(&typed_dict.values(), &DataType::Binary).unwrap();
+        let binary_dict_array =
+            DictionaryArray::<Int8Type>::new(typed_dict.keys().clone(), binary_buffer);
+        let typed_binary_dict = binary_dict_array.downcast_dict::<BinaryArray>().unwrap();
+
+        let binary_view_array = {
+            let mut builder = BinaryViewBuilder::new().with_block_size(8); // multiple buffers.
+            for v in typed_binary_dict.into_iter() {
+                builder.append_option(v);
+            }
+            builder.finish()
+        };
+        let expected_binary_array_type = binary_view_array.data_type();
+        let casted_binary_array = cast(&binary_dict_array, expected_binary_array_type).unwrap();
+        assert_eq!(casted_binary_array.data_type(), expected_binary_array_type);
+        assert_eq!(casted_binary_array.as_ref(), &binary_view_array);
+    }
+
+    #[test]
+    fn test_view_to_dict() {
+        let string_view_array = StringViewArray::from_iter(VIEW_TEST_DATA);
+        let string_dict_array: DictionaryArray<Int8Type> = VIEW_TEST_DATA.into_iter().collect();
+        let casted_type = string_dict_array.data_type();
+        let casted_dict_array = cast(&string_view_array, casted_type).unwrap();
+        assert_eq!(casted_dict_array.data_type(), casted_type);
+        assert_eq!(casted_dict_array.as_ref(), &string_dict_array);
+
+        let binary_view_array = BinaryViewArray::from_iter(VIEW_TEST_DATA);
+        let binary_dict_array = string_dict_array.downcast_dict::<StringArray>().unwrap();
+        let binary_buffer = cast(&binary_dict_array.values(), &DataType::Binary).unwrap();
+        let binary_dict_array =
+            DictionaryArray::<Int8Type>::new(binary_dict_array.keys().clone(), binary_buffer);
+        let casted_type = binary_dict_array.data_type();
+        let casted_binary_array = cast(&binary_view_array, casted_type).unwrap();
+        assert_eq!(casted_binary_array.data_type(), casted_type);
+        assert_eq!(casted_binary_array.as_ref(), &binary_dict_array);
+    }
+
     #[test]
     fn test_view_to_string() {
         _test_view_to_string::<i32>();
@@ -5271,24 +5303,15 @@ mod tests {
     where
         O: OffsetSizeTrait,
     {
-        let data: Vec<Option<&str>> = vec![
-            Some("hello"),
-            Some("world"),
-            None,
-            Some("large payload over 12 bytes"),
-            Some("lulu"),
-        ];
-
         let view_array = {
-            // ["hello", "world", null, "large payload over 12 bytes", "lulu"]
             let mut builder = StringViewBuilder::new().with_block_size(8); // multiple buffers.
-            for s in data.iter() {
+            for s in VIEW_TEST_DATA.iter() {
                 builder.append_option(*s);
             }
             builder.finish()
         };
 
-        let expected_string_array = GenericStringArray::<O>::from(data);
+        let expected_string_array = GenericStringArray::<O>::from_iter(VIEW_TEST_DATA);
         let expected_type = expected_string_array.data_type();
 
         assert!(can_cast_types(view_array.data_type(), expected_type));
@@ -5309,24 +5332,15 @@ mod tests {
     where
         O: OffsetSizeTrait,
     {
-        let data: Vec<Option<&[u8]>> = vec![
-            Some(b"hello"),
-            Some(b"world"),
-            None,
-            Some(b"large payload over 12 bytes"),
-            Some(b"lulu"),
-        ];
-
         let view_array = {
-            // ["hello", "world", null, "large payload over 12 bytes", "lulu"]
             let mut builder = BinaryViewBuilder::new().with_block_size(8); // multiple buffers.
-            for s in data.iter() {
+            for s in VIEW_TEST_DATA.iter() {
                 builder.append_option(*s);
             }
             builder.finish()
         };
 
-        let expected_binary_array = GenericBinaryArray::<O>::from(data);
+        let expected_binary_array = GenericBinaryArray::<O>::from_iter(VIEW_TEST_DATA);
         let expected_type = expected_binary_array.data_type();
 
         assert!(can_cast_types(view_array.data_type(), expected_type));
diff --git a/arrow-data/src/data.rs b/arrow-data/src/data.rs
index 5ee966394..b6ff4518e 100644
--- a/arrow-data/src/data.rs
+++ b/arrow-data/src/data.rs
@@ -24,9 +24,9 @@ use arrow_buffer::{
     bit_util, i256, ArrowNativeType, Buffer, IntervalDayTime, IntervalMonthDayNano, MutableBuffer,
 };
 use arrow_schema::{ArrowError, DataType, UnionMode};
+use std::mem;
 use std::ops::Range;
 use std::sync::Arc;
-use std::{mem, usize};
 
 use crate::{equal, validate_binary_view, validate_string_view};
 
@@ -573,6 +573,7 @@ impl ArrayData {
                 DataType::Binary | DataType::Utf8 => {
                     (vec![zeroed((len + 1) * 4), zeroed(0)], vec![], true)
                 }
+                DataType::BinaryView | DataType::Utf8View => (vec![zeroed(len * 16)], vec![], true),
                 DataType::LargeBinary | DataType::LargeUtf8 => {
                     (vec![zeroed((len + 1) * 8), zeroed(0)], vec![], true)
                 }
@@ -2209,4 +2210,20 @@ mod tests {
         data.align_buffers();
         data.validate_full().unwrap();
     }
+
+    #[test]
+    fn test_null_view_types() {
+        let array_len = 32;
+        let array = ArrayData::new_null(&DataType::BinaryView, array_len);
+        assert_eq!(array.len(), array_len);
+        for i in 0..array.len() {
+            assert!(array.is_null(i));
+        }
+
+        let array = ArrayData::new_null(&DataType::Utf8View, array_len);
+        assert_eq!(array.len(), array_len);
+        for i in 0..array.len() {
+            assert!(array.is_null(i));
+        }
+    }
 }
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index ef08a6130..f74a86e01 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -1742,7 +1742,7 @@ mod tests {
         let array1 = NullArray::new(32);
         let array2 = Int32Array::from(vec![1; 32]);
         let array3 = NullArray::new(32);
-        let array4 = Float64Array::from(vec![std::f64::NAN; 32]);
+        let array4 = Float64Array::from(vec![f64::NAN; 32]);
         let batch = RecordBatch::try_new(
             Arc::new(schema.clone()),
             vec![
diff --git a/arrow-json/src/reader/mod.rs b/arrow-json/src/reader/mod.rs
index 9a113ee2b..0a50cfac6 100644
--- a/arrow-json/src/reader/mod.rs
+++ b/arrow-json/src/reader/mod.rs
@@ -2282,8 +2282,8 @@ mod tests {
     fn test_coercing_primitive_into_string_decoder() {
         let buf = &format!(
             r#"[{{"a": 1, "b": "A", "c": "T"}}, {{"a": 2, "b": "BB", "c": "F"}}, {{"a": {}, "b": 123, "c": false}}, {{"a": {}, "b": 789, "c": true}}]"#,
-            (std::i32::MAX as i64 + 10),
-            std::i64::MAX - 10
+            (i32::MAX as i64 + 10),
+            i64::MAX - 10
         );
         let schema = Schema::new(vec![
             Field::new("a", DataType::Float64, true),
@@ -2306,8 +2306,8 @@ mod tests {
                     Arc::new(Float64Array::from(vec![
                         1.0,
                         2.0,
-                        (std::i32::MAX as i64 + 10) as f64,
-                        (std::i64::MAX - 10) as f64
+                        (i32::MAX as i64 + 10) as f64,
+                        (i64::MAX - 10) as f64
                     ])),
                     Arc::new(StringArray::from(vec!["A", "BB", "123", "789"])),
                     Arc::new(StringArray::from(vec!["T", "F", "false", "true"])),
diff --git a/arrow-schema/src/field.rs b/arrow-schema/src/field.rs
index b84a2568e..a84a6ada3 100644
--- a/arrow-schema/src/field.rs
+++ b/arrow-schema/src/field.rs
@@ -483,6 +483,30 @@ impl Field {
                     ));
                 }
             },
+            DataType::List(field) => match &from.data_type {
+                DataType::List(from_field) => {
+                    let mut f = (**field).clone();
+                    f.try_merge(from_field)?;
+                    (*field) = Arc::new(f);
+                },
+                _ => {
+                    return Err(ArrowError::SchemaError(
+                        format!("Fail to merge schema field '{}' because the from data_type = {} is not DataType::List",
+                            self.name, from.data_type)
+                ))}
+            },
+            DataType::LargeList(field) => match &from.data_type {
+                DataType::LargeList(from_field) => {
+                    let mut f = (**field).clone();
+                    f.try_merge(from_field)?;
+                    (*field) = Arc::new(f);
+                },
+                _ => {
+                    return Err(ArrowError::SchemaError(
+                        format!("Fail to merge schema field '{}' because the from data_type = {} is not DataType::LargeList",
+                            self.name, from.data_type)
+                ))}
+            },
             DataType::Null => {
                 self.nullable = true;
                 self.data_type = from.data_type.clone();
@@ -509,9 +533,7 @@ impl Field {
             | DataType::LargeBinary
             | DataType::BinaryView
             | DataType::Interval(_)
-            | DataType::LargeList(_)
             | DataType::LargeListView(_)
-            | DataType::List(_)
             | DataType::ListView(_)
             | DataType::Map(_, _)
             | DataType::Dictionary(_, _)
@@ -623,6 +645,93 @@ mod test {
         assert_eq!(Field::new("c2", DataType::Utf8, true), field2);
     }
 
+    #[test]
+    fn test_merge_with_nested_null() {
+        let mut struct1 = Field::new(
+            "s1",
+            DataType::Struct(Fields::from(vec![Field::new(
+                "inner",
+                DataType::Float32,
+                false,
+            )])),
+            false,
+        );
+
+        let struct2 = Field::new(
+            "s2",
+            DataType::Struct(Fields::from(vec![Field::new(
+                "inner",
+                DataType::Null,
+                false,
+            )])),
+            true,
+        );
+
+        struct1
+            .try_merge(&struct2)
+            .expect("should widen inner field's type to nullable float");
+        assert_eq!(
+            Field::new(
+                "s1",
+                DataType::Struct(Fields::from(vec![Field::new(
+                    "inner",
+                    DataType::Float32,
+                    true,
+                )])),
+                true,
+            ),
+            struct1
+        );
+
+        let mut list1 = Field::new(
+            "l1",
+            DataType::List(Field::new("inner", DataType::Float32, false).into()),
+            false,
+        );
+
+        let list2 = Field::new(
+            "l2",
+            DataType::List(Field::new("inner", DataType::Null, false).into()),
+            true,
+        );
+
+        list1
+            .try_merge(&list2)
+            .expect("should widen inner field's type to nullable float");
+        assert_eq!(
+            Field::new(
+                "l1",
+                DataType::List(Field::new("inner", DataType::Float32, true).into()),
+                true,
+            ),
+            list1
+        );
+
+        let mut large_list1 = Field::new(
+            "ll1",
+            DataType::LargeList(Field::new("inner", DataType::Float32, false).into()),
+            false,
+        );
+
+        let large_list2 = Field::new(
+            "ll2",
+            DataType::LargeList(Field::new("inner", DataType::Null, false).into()),
+            true,
+        );
+
+        large_list1
+            .try_merge(&large_list2)
+            .expect("should widen inner field's type to nullable float");
+        assert_eq!(
+            Field::new(
+                "ll1",
+                DataType::LargeList(Field::new("inner", DataType::Float32, true).into()),
+                true,
+            ),
+            large_list1
+        );
+    }
+
     #[test]
     fn test_fields_with_dict_id() {
         let dict1 = Field::new_dict(
diff --git a/arrow-select/src/filter.rs b/arrow-select/src/filter.rs
index 8e06b07f5..65ccbe1e0 100644
--- a/arrow-select/src/filter.rs
+++ b/arrow-select/src/filter.rs
@@ -552,7 +552,10 @@ fn filter_native<T: ArrowNativeType>(values: &[T], predicate: &FilterPredicate)
 }
 
 /// `filter` implementation for primitive arrays
-fn filter_primitive<T>(array: &PrimitiveArray<T>, predicate: &FilterPredicate) -> PrimitiveArray<T>
+pub(crate) fn filter_primitive<T>(
+    array: &PrimitiveArray<T>,
+    predicate: &FilterPredicate,
+) -> PrimitiveArray<T>
 where
     T: ArrowPrimitiveType,
 {
diff --git a/arrow-select/src/take.rs b/arrow-select/src/take.rs
index b8d59142d..d6892eb0a 100644
--- a/arrow-select/src/take.rs
+++ b/arrow-select/src/take.rs
@@ -31,6 +31,8 @@ use arrow_schema::{ArrowError, DataType, FieldRef, UnionMode};
 
 use num::{One, Zero};
 
+use crate::filter::{filter_primitive, FilterBuilder};
+
 /// Take elements by index from [Array], creating a new [Array] from those indexes.
 ///
 /// ```text
@@ -240,6 +242,44 @@ fn take_impl<IndexType: ArrowPrimitiveType>(
             let array = UnionArray::try_new(fields.clone(), type_ids, None, children)?;
             Ok(Arc::new(array))
         }
+        DataType::Union(fields, UnionMode::Dense) => {
+            let values = values.as_any().downcast_ref::<UnionArray>().unwrap();
+
+            let type_ids = <PrimitiveArray<Int8Type>>::new(take_native(values.type_ids(), indices), None);
+            let offsets = <PrimitiveArray<Int32Type>>::new(take_native(values.offsets().unwrap(), indices), None);
+
+            let children = fields.iter()
+                .map(|(field_type_id, _)| {
+                    let mask = BooleanArray::from_unary(&type_ids, |value_type_id| value_type_id == field_type_id);
+                    let predicate = FilterBuilder::new(&mask).build();
+
+                    let indices = filter_primitive(&offsets, &predicate);
+
+                    let values = values.child(field_type_id);
+
+                    take_impl(values, &indices)
+                })
+                .collect::<Result<_, _>>()?;
+
+            let mut child_offsets = [0; 128];
+
+            let offsets = type_ids.values()
+                .iter()
+                .map(|&i| {
+                    let offset = child_offsets[i as usize];
+
+                    child_offsets[i as usize] += 1;
+
+                    offset
+                })
+                .collect();
+
+            let (_, type_ids, _) = type_ids.into_parts();
+
+            let array = UnionArray::try_new(fields.clone(), type_ids, Some(offsets), children)?;
+
+            Ok(Arc::new(array))
+        }
         t => unimplemented!("Take not supported for data type {:?}", t)
     }
 }
@@ -2146,7 +2186,7 @@ mod tests {
     }
 
     #[test]
-    fn test_take_union() {
+    fn test_take_union_sparse() {
         let structs = create_test_struct(vec![
             Some((Some(true), Some(42))),
             Some((Some(false), Some(28))),
@@ -2183,4 +2223,90 @@ mod tests {
         let expected = vec![Some("a"), None, None, Some("a"), Some("c"), Some("d")];
         assert_eq!(expected, actual);
     }
+
+    #[test]
+    fn test_take_union_dense() {
+        let type_ids = vec![0, 1, 1, 0, 0, 1, 0];
+        let offsets = vec![0, 0, 1, 1, 2, 2, 3];
+        let ints = vec![10, 20, 30, 40];
+        let strings = vec![Some("a"), None, Some("c"), Some("d")];
+
+        let indices = vec![0, 3, 1, 0, 2, 4];
+
+        let taken_type_ids = vec![0, 0, 1, 0, 1, 0];
+        let taken_offsets = vec![0, 1, 0, 2, 1, 3];
+        let taken_ints = vec![10, 20, 10, 30];
+        let taken_strings = vec![Some("a"), None];
+
+        let type_ids = <ScalarBuffer<i8>>::from(type_ids);
+        let offsets = <ScalarBuffer<i32>>::from(offsets);
+        let ints = UInt32Array::from(ints);
+        let strings = StringArray::from(strings);
+
+        let union_fields = [
+            (
+                0,
+                Arc::new(Field::new("f1", ints.data_type().clone(), true)),
+            ),
+            (
+                1,
+                Arc::new(Field::new("f2", strings.data_type().clone(), true)),
+            ),
+        ]
+        .into_iter()
+        .collect();
+
+        let array = UnionArray::try_new(
+            union_fields,
+            type_ids,
+            Some(offsets),
+            vec![Arc::new(ints), Arc::new(strings)],
+        )
+        .unwrap();
+
+        let index = UInt32Array::from(indices);
+
+        let actual = take(&array, &index, None).unwrap();
+        let actual = actual.as_any().downcast_ref::<UnionArray>().unwrap();
+
+        assert_eq!(actual.offsets(), Some(&ScalarBuffer::from(taken_offsets)));
+        assert_eq!(actual.type_ids(), &ScalarBuffer::from(taken_type_ids));
+        assert_eq!(
+            UInt32Array::from(actual.child(0).to_data()),
+            UInt32Array::from(taken_ints)
+        );
+        assert_eq!(
+            StringArray::from(actual.child(1).to_data()),
+            StringArray::from(taken_strings)
+        );
+    }
+
+    #[test]
+    fn test_take_union_dense_using_builder() {
+        let mut builder = UnionBuilder::new_dense();
+
+        builder.append::<Int32Type>("a", 1).unwrap();
+        builder.append::<Float64Type>("b", 3.0).unwrap();
+        builder.append::<Int32Type>("a", 4).unwrap();
+        builder.append::<Int32Type>("a", 5).unwrap();
+        builder.append::<Float64Type>("b", 2.0).unwrap();
+
+        let union = builder.build().unwrap();
+
+        let indices = UInt32Array::from(vec![2, 0, 1, 2]);
+
+        let mut builder = UnionBuilder::new_dense();
+
+        builder.append::<Int32Type>("a", 4).unwrap();
+        builder.append::<Int32Type>("a", 1).unwrap();
+        builder.append::<Float64Type>("b", 3.0).unwrap();
+        builder.append::<Int32Type>("a", 4).unwrap();
+
+        let taken = builder.build().unwrap();
+
+        assert_eq!(
+            taken.to_data(),
+            take(&union, &indices, None).unwrap().to_data()
+        );
+    }
 }
diff --git a/arrow/benches/cast_kernels.rs b/arrow/benches/cast_kernels.rs
index 228408e57..8803e8eea 100644
--- a/arrow/benches/cast_kernels.rs
+++ b/arrow/benches/cast_kernels.rs
@@ -114,6 +114,17 @@ fn build_decimal256_array(size: usize, precision: u8, scale: i8) -> ArrayRef {
     )
 }
 
+fn build_dict_array(size: usize) -> ArrayRef {
+    let values = StringArray::from_iter([
+        Some("small"),
+        Some("larger string more than 12 bytes"),
+        None,
+    ]);
+    let keys = UInt64Array::from_iter((0..size as u64).map(|v| v % 3));
+
+    Arc::new(DictionaryArray::new(keys, Arc::new(values)))
+}
+
 // cast array from specified primitive array type to desired data type
 fn cast_array(array: &ArrayRef, to_type: DataType) {
     criterion::black_box(cast(array, &to_type).unwrap());
@@ -138,6 +149,9 @@ fn add_benchmark(c: &mut Criterion) {
     let decimal128_array = build_decimal128_array(512, 10, 3);
     let decimal256_array = build_decimal256_array(512, 50, 3);
 
+    let dict_array = build_dict_array(10_000);
+    let string_view_array = cast(&dict_array, &DataType::Utf8View).unwrap();
+
     c.bench_function("cast int32 to int32 512", |b| {
         b.iter(|| cast_array(&i32_array, DataType::Int32))
     });
@@ -237,6 +251,17 @@ fn add_benchmark(c: &mut Criterion) {
     c.bench_function("cast decimal256 to decimal256 512 with same scale", |b| {
         b.iter(|| cast_array(&decimal256_array, DataType::Decimal256(60, 3)))
     });
+    c.bench_function("cast dict to string view", |b| {
+        b.iter(|| cast_array(&dict_array, DataType::Utf8View))
+    });
+    c.bench_function("cast string view to dict", |b| {
+        b.iter(|| {
+            cast_array(
+                &string_view_array,
+                DataType::Dictionary(Box::new(DataType::UInt64), Box::new(DataType::Utf8)),
+            )
+        })
+    });
 }
 
 criterion_group!(benches, add_benchmark);
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index 1e42093fd..ffef3fb33 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -898,7 +898,6 @@ impl AmazonS3Builder {
             info!("Using Instance credential provider");
 
             let token = InstanceCredentialProvider {
-                cache: Default::default(),
                 imdsv1_fallback: self.imdsv1_fallback.get()?,
                 metadata_endpoint: self
                     .metadata_endpoint
diff --git a/object_store/src/aws/credential.rs b/object_store/src/aws/credential.rs
index 08831fd51..c13f8aafc 100644
--- a/object_store/src/aws/credential.rs
+++ b/object_store/src/aws/credential.rs
@@ -424,7 +424,6 @@ fn canonicalize_headers(header_map: &HeaderMap) -> (String, String) {
 /// <https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html>
 #[derive(Debug)]
 pub struct InstanceCredentialProvider {
-    pub cache: TokenCache<Arc<AwsCredential>>,
     pub imdsv1_fallback: bool,
     pub metadata_endpoint: String,
 }
diff --git a/object_store/src/buffered.rs b/object_store/src/buffered.rs
index feb84d4d0..c7b71aa1c 100644
--- a/object_store/src/buffered.rs
+++ b/object_store/src/buffered.rs
@@ -238,11 +238,11 @@ enum BufWriterState {
     /// Buffer up to capacity bytes
     Buffer(Path, PutPayloadMut),
     /// [`ObjectStore::put_multipart`]
-    Prepare(BoxFuture<'static, std::io::Result<WriteMultipart>>),
+    Prepare(BoxFuture<'static, crate::Result<WriteMultipart>>),
     /// Write to a multipart upload
     Write(Option<WriteMultipart>),
     /// [`ObjectStore::put`]
-    Flush(BoxFuture<'static, std::io::Result<()>>),
+    Flush(BoxFuture<'static, crate::Result<()>>),
 }
 
 impl BufWriter {
@@ -289,6 +289,58 @@ impl BufWriter {
         }
     }
 
+    /// Write data to the writer in [`Bytes`].
+    ///
+    /// Unlike [`AsyncWrite::poll_write`], `put` can write data without extra copying.
+    ///
+    /// This API is recommended while the data source generates [`Bytes`].
+    pub async fn put(&mut self, bytes: Bytes) -> crate::Result<()> {
+        loop {
+            return match &mut self.state {
+                BufWriterState::Write(Some(write)) => {
+                    write.wait_for_capacity(self.max_concurrency).await?;
+                    write.put(bytes);
+                    Ok(())
+                }
+                BufWriterState::Write(None) | BufWriterState::Flush(_) => {
+                    panic!("Already shut down")
+                }
+                // NOTE
+                //
+                // This case should never happen in practice, but rust async API does
+                // make it possible for users to call `put` before `poll_write` returns `Ready`.
+                //
+                // We allow such usage by `await` the future and continue the loop.
+                BufWriterState::Prepare(f) => {
+                    self.state = BufWriterState::Write(f.await?.into());
+                    continue;
+                }
+                BufWriterState::Buffer(path, b) => {
+                    if b.content_length().saturating_add(bytes.len()) < self.capacity {
+                        b.push(bytes);
+                        Ok(())
+                    } else {
+                        let buffer = std::mem::take(b);
+                        let path = std::mem::take(path);
+                        let opts = PutMultipartOpts {
+                            attributes: self.attributes.take().unwrap_or_default(),
+                            tags: self.tags.take().unwrap_or_default(),
+                        };
+                        let upload = self.store.put_multipart_opts(&path, opts).await?;
+                        let mut chunked =
+                            WriteMultipart::new_with_chunk_size(upload, self.capacity);
+                        for chunk in buffer.freeze() {
+                            chunked.put(chunk);
+                        }
+                        chunked.put(bytes);
+                        self.state = BufWriterState::Write(Some(chunked));
+                        Ok(())
+                    }
+                }
+            };
+        }
+    }
+
     /// Abort this writer, cleaning up any partially uploaded state
     ///
     /// # Panic
@@ -384,7 +436,7 @@ impl AsyncWrite for BufWriter {
                         Ok(())
                     }));
                 }
-                BufWriterState::Flush(f) => return f.poll_unpin(cx),
+                BufWriterState::Flush(f) => return f.poll_unpin(cx).map_err(std::io::Error::from),
                 BufWriterState::Write(x) => {
                     let upload = x.take().unwrap();
                     self.state = BufWriterState::Flush(
@@ -416,6 +468,7 @@ mod tests {
     use crate::memory::InMemory;
     use crate::path::Path;
     use crate::{Attribute, GetOptions};
+    use itertools::Itertools;
     use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 
     #[tokio::test]
@@ -547,4 +600,58 @@ mod tests {
         assert_eq!(response.meta.size, 40);
         assert_eq!(response.attributes, attributes);
     }
+
+    #[tokio::test]
+    async fn test_buf_writer_with_put() {
+        let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+        let path = Path::from("file.txt");
+
+        // Test put
+        let mut writer = BufWriter::with_capacity(Arc::clone(&store), path.clone(), 30);
+        writer
+            .put(Bytes::from((0..20).collect_vec()))
+            .await
+            .unwrap();
+        writer
+            .put(Bytes::from((20..25).collect_vec()))
+            .await
+            .unwrap();
+        writer.shutdown().await.unwrap();
+        let response = store
+            .get_opts(
+                &path,
+                GetOptions {
+                    head: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.meta.size, 25);
+        assert_eq!(response.bytes().await.unwrap(), (0..25).collect_vec());
+
+        // Test multipart
+        let mut writer = BufWriter::with_capacity(Arc::clone(&store), path.clone(), 30);
+        writer
+            .put(Bytes::from((0..20).collect_vec()))
+            .await
+            .unwrap();
+        writer
+            .put(Bytes::from((20..40).collect_vec()))
+            .await
+            .unwrap();
+        writer.shutdown().await.unwrap();
+        let response = store
+            .get_opts(
+                &path,
+                GetOptions {
+                    head: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.meta.size, 40);
+        assert_eq!(response.bytes().await.unwrap(), (0..40).collect_vec());
+    }
 }
diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs
index 6b95324be..070dda97c 100644
--- a/parquet/src/arrow/arrow_reader/mod.rs
+++ b/parquet/src/arrow/arrow_reader/mod.rs
@@ -248,7 +248,9 @@ impl<T> ArrowReaderBuilder<T> {
 /// is then read from the file, including projection and filter pushdown
 #[derive(Debug, Clone, Default)]
 pub struct ArrowReaderOptions {
+    /// Should the reader strip any user defined metadata from the Arrow schema
     skip_arrow_metadata: bool,
+    /// If true, attempt to read `OffsetIndex` and `ColumnIndex`
     pub(crate) page_index: bool,
 }
 
@@ -271,34 +273,58 @@ impl ArrowReaderOptions {
         }
     }
 
-    /// Enable decoding of the [`PageIndex`], if present (defaults to `false`)
+    /// Enable reading [`PageIndex`], if present (defaults to `false`)
     ///
     /// The `PageIndex` can be used to push down predicates to the parquet scan,
     /// potentially eliminating unnecessary IO, by some query engines.
     ///
+    /// If this is enabled, [`ParquetMetaData::column_index`] and
+    /// [`ParquetMetaData::offset_index`] will be populated if the corresponding
+    /// information is present in the file.
+    ///
     /// [`PageIndex`]: https://github.com/apache/parquet-format/blob/master/PageIndex.md
+    /// [`ParquetMetaData::column_index`]: crate::file::metadata::ParquetMetaData::column_index
+    /// [`ParquetMetaData::offset_index`]: crate::file::metadata::ParquetMetaData::offset_index
     pub fn with_page_index(self, page_index: bool) -> Self {
         Self { page_index, ..self }
     }
 }
 
-/// The cheaply clone-able metadata necessary to construct a [`ArrowReaderBuilder`]
+/// The metadata necessary to construct a [`ArrowReaderBuilder`]
+///
+/// Note this structure is cheaply clone-able as it consists of several arcs.
+///
+/// This structure allows
+///
+/// 1. Loading metadata for a file once and then using that same metadata to
+/// construct multiple separate readers, for example, to distribute readers
+/// across multiple threads
 ///
-/// This allows loading the metadata for a file once and then using this to construct
-/// multiple separate readers, for example, to distribute readers across multiple threads
+/// 2. Using a cached copy of the [`ParquetMetadata`] rather than reading it
+/// from the file each time a reader is constructed.
+///
+/// [`ParquetMetadata`]: crate::file::metadata::ParquetMetaData
 #[derive(Debug, Clone)]
 pub struct ArrowReaderMetadata {
+    /// The Parquet Metadata, if known aprior
     pub(crate) metadata: Arc<ParquetMetaData>,
-
+    /// The Arrow Schema
     pub(crate) schema: SchemaRef,
 
     pub(crate) fields: Option<Arc<ParquetField>>,
 }
 
 impl ArrowReaderMetadata {
-    /// Loads [`ArrowReaderMetadata`] from the provided [`ChunkReader`]
+    /// Loads [`ArrowReaderMetadata`] from the provided [`ChunkReader`], if necessary
+    ///
+    /// See [`ParquetRecordBatchReaderBuilder::new_with_metadata`] for an
+    /// example of how this can be used
+    ///
+    /// # Notes
     ///
-    /// See [`ParquetRecordBatchReaderBuilder::new_with_metadata`] for how this can be used
+    /// If `options` has [`ArrowReaderOptions::with_page_index`] true, but
+    /// `Self::metadata` is missing the page index, this function will attempt
+    /// to load the page index by making an object store request.
     pub fn load<T: ChunkReader>(reader: &T, options: ArrowReaderOptions) -> Result<Self> {
         let mut metadata = footer::parse_metadata(reader)?;
         if options.page_index {
@@ -320,6 +346,12 @@ impl ArrowReaderMetadata {
         Self::try_new(Arc::new(metadata), options)
     }
 
+    /// Create a new [`ArrowReaderMetadata`]
+    ///
+    /// # Notes
+    ///
+    /// This function does not attempt to load the PageIndex if not present in the metadata.
+    /// See [`Self::load`] for more details.
     pub fn try_new(metadata: Arc<ParquetMetaData>, options: ArrowReaderOptions) -> Result<Self> {
         let kv_metadata = match options.skip_arrow_metadata {
             true => None,
@@ -407,9 +439,17 @@ impl<T: ChunkReader + 'static> ParquetRecordBatchReaderBuilder<T> {
 
     /// Create a [`ParquetRecordBatchReaderBuilder`] from the provided [`ArrowReaderMetadata`]
     ///
-    /// This allows loading metadata once and using it to create multiple builders with
-    /// potentially different settings
+    /// This interface allows:
     ///
+    /// 1. Loading metadata once and using it to create multiple builders with
+    /// potentially different settings or run on different threads
+    ///
+    /// 2. Using a cached copy of the metadata rather than re-reading it from the
+    /// file each time a reader is constructed.
+    ///
+    /// See the docs on [`ArrowReaderMetadata`] for more details
+    ///
+    /// # Example
     /// ```
     /// # use std::fs::metadata;
     /// # use std::sync::Arc;
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
index 53287dec5..0beb93f80 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -107,6 +107,20 @@ mod levels;
 /// }
 /// ```
 ///
+/// ## Type Support
+///
+/// The writer supports writing all Arrow [`DataType`]s that have a direct mapping to
+/// Parquet types including  [`StructArray`] and [`ListArray`].
+///
+/// The following are not supported:
+///
+/// * [`IntervalMonthDayNanoArray`]: Parquet does not [support nanosecond intervals].
+///
+/// [`DataType`]: https://docs.rs/arrow/latest/arrow/datatypes/enum.DataType.html
+/// [`StructArray`]: https://docs.rs/arrow/latest/arrow/array/struct.StructArray.html
+/// [`ListArray`]: https://docs.rs/arrow/latest/arrow/array/type.ListArray.html
+/// [`IntervalMonthDayNanoArray`]: https://docs.rs/arrow/latest/arrow/array/type.IntervalMonthDayNanoArray.html
+/// [support nanosecond intervals]: https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#interval
 pub struct ArrowWriter<W: Write> {
     /// Underlying Parquet writer
     writer: SerializedFileWriter<W>,
@@ -2103,6 +2117,16 @@ mod tests {
         values_required::<BinaryArray, _>(many_vecs_iter);
     }
 
+    #[test]
+    fn binary_view_single_column() {
+        let one_vec: Vec<u8> = (0..SMALL_SIZE as u8).collect();
+        let many_vecs: Vec<_> = std::iter::repeat(one_vec).take(SMALL_SIZE).collect();
+        let many_vecs_iter = many_vecs.iter().map(|v| v.as_slice());
+
+        // BinaryArrays can't be built from Vec<Option<&str>>, so only call `values_required`
+        values_required::<BinaryViewArray, _>(many_vecs_iter);
+    }
+
     #[test]
     fn i32_column_bloom_filter() {
         let array = Arc::new(Int32Array::from_iter(0..SMALL_SIZE as i32));
@@ -2195,6 +2219,14 @@ mod tests {
         required_and_optional::<LargeStringArray, _>(raw_strs);
     }
 
+    #[test]
+    fn string_view_single_column() {
+        let raw_values: Vec<_> = (0..SMALL_SIZE).map(|i| i.to_string()).collect();
+        let raw_strs = raw_values.iter().map(|s| s.as_str());
+
+        required_and_optional::<StringViewArray, _>(raw_strs);
+    }
+
     #[test]
     fn null_list_single_column() {
         let null_field = Field::new("item", DataType::Null, true);
diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs
index 1e298c654..0a72583b9 100644
--- a/parquet/src/arrow/async_reader/mod.rs
+++ b/parquet/src/arrow/async_reader/mod.rs
@@ -202,6 +202,12 @@ impl ArrowReaderMetadata {
     /// Returns a new [`ArrowReaderMetadata`] for this builder
     ///
     /// See [`ParquetRecordBatchStreamBuilder::new_with_metadata`] for how this can be used
+    ///
+    /// # Notes
+    ///
+    /// If `options` has [`ArrowReaderOptions::with_page_index`] true, but
+    /// `Self::metadata` is missing the page index, this function will attempt
+    /// to load the page index by making an object store request.
     pub async fn load_async<T: AsyncFileReader>(
         input: &mut T,
         options: ArrowReaderOptions,
diff --git a/parquet/src/arrow/async_reader/store.rs b/parquet/src/arrow/async_reader/store.rs
index 293b91aea..77c00e91a 100644
--- a/parquet/src/arrow/async_reader/store.rs
+++ b/parquet/src/arrow/async_reader/store.rs
@@ -25,7 +25,7 @@ use futures::{FutureExt, TryFutureExt};
 use object_store::{ObjectMeta, ObjectStore};
 
 use crate::arrow::async_reader::{AsyncFileReader, MetadataLoader};
-use crate::errors::{ParquetError, Result};
+use crate::errors::Result;
 use crate::file::metadata::ParquetMetaData;
 
 /// Reads Parquet files in object storage using [`ObjectStore`].
@@ -105,7 +105,7 @@ impl AsyncFileReader for ParquetObjectReader {
     fn get_bytes(&mut self, range: Range<usize>) -> BoxFuture<'_, Result<Bytes>> {
         self.store
             .get_range(&self.meta.location, range)
-            .map_err(|e| ParquetError::General(format!("AsyncChunkReader::get_bytes error: {e}")))
+            .map_err(|e| e.into())
             .boxed()
     }
 
@@ -117,11 +117,7 @@ impl AsyncFileReader for ParquetObjectReader {
             self.store
                 .get_ranges(&self.meta.location, &ranges)
                 .await
-                .map_err(|e| {
-                    ParquetError::General(format!(
-                        "ParquetObjectReader::get_byte_ranges error: {e}"
-                    ))
-                })
+                .map_err(|e| e.into())
         }
         .boxed()
     }
diff --git a/parquet/src/arrow/buffer/offset_buffer.rs b/parquet/src/arrow/buffer/offset_buffer.rs
index 32bb9d086..806f144d9 100644
--- a/parquet/src/arrow/buffer/offset_buffer.rs
+++ b/parquet/src/arrow/buffer/offset_buffer.rs
@@ -156,16 +156,20 @@ impl<I: OffsetSizeTrait> OffsetBuffer<I> {
 
     fn build_generic_byte_view(self) -> GenericByteViewBuilder<BinaryViewType> {
         let mut builder = GenericByteViewBuilder::<BinaryViewType>::with_capacity(self.len());
-        let mut values = self.values;
+        let buffer = self.values.into();
+        let block = builder.append_block(buffer);
         for window in self.offsets.windows(2) {
             let start = window[0];
             let end = window[1];
             let len = (end - start).to_usize().unwrap();
-            let b = values.drain(..len).collect::<Vec<u8>>();
-            if b.is_empty() {
-                builder.append_null();
+
+            if len != 0 {
+                // Safety: (1) the buffer is valid (2) the offsets are valid (3) the values in between are of ByteViewType
+                unsafe {
+                    builder.append_view_unchecked(block, start.as_usize() as u32, len as u32);
+                }
             } else {
-                builder.append_value(b);
+                builder.append_null();
             }
         }
         builder
diff --git a/parquet/src/column/reader.rs b/parquet/src/column/reader.rs
index 53d56dd6b..b40ca2b78 100644
--- a/parquet/src/column/reader.rs
+++ b/parquet/src/column/reader.rs
@@ -672,8 +672,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
     test!(
         test_read_plain_v2_int32,
@@ -684,8 +684,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
 
     test!(
@@ -697,8 +697,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         17,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
     test!(
         test_read_plain_v2_int32_uneven,
@@ -709,8 +709,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         17,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
 
     test!(
@@ -722,8 +722,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         512,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
     test!(
         test_read_plain_v2_int32_multi_page,
@@ -734,8 +734,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         512,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
 
     // test cases when column descriptor has MAX_DEF_LEVEL = 0 and MAX_REP_LEVEL = 0
@@ -748,8 +748,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
     test!(
         test_read_plain_v2_int32_required_non_repeated,
@@ -760,8 +760,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i32::MIN,
-        std::i32::MAX
+        i32::MIN,
+        i32::MAX
     );
 
     test!(
@@ -773,8 +773,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
     test!(
         test_read_plain_v2_int64,
@@ -785,8 +785,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
 
     test!(
@@ -798,8 +798,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         17,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
     test!(
         test_read_plain_v2_int64_uneven,
@@ -810,8 +810,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         17,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
 
     test!(
@@ -823,8 +823,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         512,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
     test!(
         test_read_plain_v2_int64_multi_page,
@@ -835,8 +835,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         512,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
 
     // test cases when column descriptor has MAX_DEF_LEVEL = 0 and MAX_REP_LEVEL = 0
@@ -849,8 +849,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
     test!(
         test_read_plain_v2_int64_required_non_repeated,
@@ -861,8 +861,8 @@ mod tests {
         NUM_PAGES,
         NUM_LEVELS,
         16,
-        std::i64::MIN,
-        std::i64::MAX
+        i64::MIN,
+        i64::MAX
     );
 
     test!(
@@ -1037,8 +1037,8 @@ mod tests {
             num_pages,
             num_levels,
             batch_size,
-            std::i32::MIN,
-            std::i32::MAX,
+            i32::MIN,
+            i32::MAX,
             false,
         );
     }
diff --git a/parquet/src/encodings/decoding.rs b/parquet/src/encodings/decoding.rs
index a1924f72e..b3dd13523 100644
--- a/parquet/src/encodings/decoding.rs
+++ b/parquet/src/encodings/decoding.rs
@@ -1624,14 +1624,14 @@ mod tests {
     #[test]
     fn test_delta_bit_packed_int64_min_max() {
         let block_data = vec![
-            i64::min_value(),
-            i64::max_value(),
-            i64::min_value(),
-            i64::max_value(),
-            i64::min_value(),
-            i64::max_value(),
-            i64::min_value(),
-            i64::max_value(),
+            i64::MIN,
+            i64::MAX,
+            i64::MIN,
+            i64::MAX,
+            i64::MIN,
+            i64::MAX,
+            i64::MIN,
+            i64::MAX,
         ];
         test_delta_bit_packed_decode::<Int64Type>(vec![block_data]);
     }
diff --git a/parquet/src/errors.rs b/parquet/src/errors.rs
index 0b70266b3..a242c9768 100644
--- a/parquet/src/errors.rs
+++ b/parquet/src/errors.rs
@@ -112,6 +112,13 @@ impl From<ArrowError> for ParquetError {
     }
 }
 
+#[cfg(feature = "object_store")]
+impl From<object_store::Error> for ParquetError {
+    fn from(e: object_store::Error) -> ParquetError {
+        ParquetError::External(Box::new(e))
+    }
+}
+
 /// A specialized `Result` for Parquet errors.
 pub type Result<T, E = ParquetError> = result::Result<T, E>;
 
diff --git a/parquet/src/file/metadata.rs b/parquet/src/file/metadata.rs
index 853d5ffec..fb8f798fd 100644
--- a/parquet/src/file/metadata.rs
+++ b/parquet/src/file/metadata.rs
@@ -29,7 +29,6 @@
 //! * [`ColumnChunkMetaData`]: Metadata for each column chunk (primitive leaf)
 //!   within a Row Group including encoding and compression information,
 //!   number of values, statistics, etc.
-
 use std::ops::Range;
 use std::sync::Arc;
 
@@ -69,20 +68,20 @@ pub type ParquetColumnIndex = Vec<Vec<Index>>;
 /// parquet file.
 pub type ParquetOffsetIndex = Vec<Vec<Vec<PageLocation>>>;
 
-/// Global Parquet metadata, including [`FileMetaData`], [`RowGroupMetaData`].
+/// Parsed metadata for a single Parquet file
 ///
 /// This structure is stored in the footer of Parquet files, in the format
-/// defined by [`parquet.thrift`]. It contains:
-///
-/// * File level metadata: [`FileMetaData`]
-/// * Row Group level metadata: [`RowGroupMetaData`]
-/// * (Optional) "Page Index" structures: [`ParquetColumnIndex`] and [`ParquetOffsetIndex`]
+/// defined by [`parquet.thrift`].
 ///
-/// [`parquet.thrift`]: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift
+/// # Overview
+/// * [`FileMetaData`]: Information about the overall file (such as the schema) (See [`Self::file_metadata`])
+/// * [`RowGroupMetaData`]: Information about each Row Group (see [`Self::row_groups`])
+/// * [`ParquetColumnIndex`] and [`ParquetOffsetIndex`]: Optional "Page Index" structures (see [`Self::column_index`] and [`Self::offset_index`])
 ///
 /// This structure is read by the various readers in this crate or can be read
 /// directly from a file using the [`parse_metadata`] function.
 ///
+/// [`parquet.thrift`]: https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift
 /// [`parse_metadata`]: crate::file::footer::parse_metadata
 #[derive(Debug, Clone)]
 pub struct ParquetMetaData {
@@ -152,6 +151,11 @@ impl ParquetMetaData {
     }
 
     /// Returns the column index for this file if loaded
+    ///
+    /// Returns `None` if the parquet file does not have a `ColumnIndex` or
+    /// [ArrowReaderOptions::with_page_index] was set to false.
+    ///
+    /// [ArrowReaderOptions::with_page_index]: https://docs.rs/parquet/latest/parquet/arrow/arrow_reader/struct.ArrowReaderOptions.html#method.with_page_index
     pub fn column_index(&self) -> Option<&ParquetColumnIndex> {
         self.column_index.as_ref()
     }
@@ -162,7 +166,12 @@ impl ParquetMetaData {
         self.offset_index.as_ref()
     }
 
-    /// Returns offset indexes in this file.
+    /// Returns offset indexes in this file, if loaded
+    ///
+    /// Returns `None` if the parquet file does not have a `OffsetIndex` or
+    /// [ArrowReaderOptions::with_page_index] was set to false.
+    ///
+    /// [ArrowReaderOptions::with_page_index]: https://docs.rs/parquet/latest/parquet/arrow/arrow_reader/struct.ArrowReaderOptions.html#method.with_page_index
     pub fn offset_index(&self) -> Option<&ParquetOffsetIndex> {
         self.offset_index.as_ref()
     }
diff --git a/parquet/src/file/page_index/index_reader.rs b/parquet/src/file/page_index/index_reader.rs
index f298601f5..2ddf826fb 100644
--- a/parquet/src/file/page_index/index_reader.rs
+++ b/parquet/src/file/page_index/index_reader.rs
@@ -73,8 +73,8 @@ pub fn read_columns_indexes<R: ChunkReader>(
         .collect()
 }
 
-/// Reads per-page [`PageLocation`] for all columns of a row group by
-/// decoding the [`OffsetIndex`].
+/// Reads [`OffsetIndex`],  per-page [`PageLocation`] for all columns of a row
+/// group.
 ///
 /// Returns a vector of `location[column_number][page_number]`
 ///


Test patch:
diff --git a/parquet_derive_test/src/lib.rs b/parquet_derive_test/src/lib.rs
index 3743c6b55..b8e6c27f5 100644
--- a/parquet_derive_test/src/lib.rs
+++ b/parquet_derive_test/src/lib.rs
@@ -141,9 +141,9 @@ mod tests {
             maybe_usize: Some(4456),
             isize: -365,
             float: 3.5,
-            double: std::f64::NAN,
+            double: f64::NAN,
             maybe_float: None,
-            maybe_double: Some(std::f64::MAX),
+            maybe_double: Some(f64::MAX),
             borrowed_maybe_a_string: &maybe_a_string,
             borrowed_maybe_a_str: &maybe_a_str,
             now: chrono::Utc::now().naive_local(),
@@ -180,7 +180,7 @@ mod tests {
             u64: 4563424,
             isize: -365,
             float: 3.5,
-            double: std::f64::NAN,
+            double: f64::NAN,
             now: chrono::Utc::now().naive_local(),
             date: chrono::naive::NaiveDate::from_ymd_opt(2015, 3, 14).unwrap(),
             uuid: uuid::Uuid::new_v4(),