Repository: apache/arrow-rs
Pull request #5664 (2024-04-18 16:45:58)

Add `ListView` & `LargeListView` basic construction and validation
# 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.
-->

A part of #5501 

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

# What changes are included in this PR?

Add ListView & LargeListView basic ArrayData construction and validation 

<!--
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/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 193f30d85..1ce2c1301 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -95,6 +95,11 @@ jobs:
         with:
           path: rust
           fetch-depth: 0
+      # Workaround https://github.com/rust-lang/jobserver-rs/issues/87
+      # Can be removed once https://github.com/rust-lang/jobserver-rs/pull/88 is released
+      - name: Downgrade jobserver
+        working-directory: rust
+        run: cargo update -p cc --precise 1.0.94 && cargo update -p jobserver --precise 0.1.28
       - name: Build
         run: conda run --no-capture-output ci/scripts/integration_arrow_build.sh $PWD /build
       - name: Run
diff --git a/arrow-array/Cargo.toml b/arrow-array/Cargo.toml
index 0bc2facb5..b00d2c88e 100644
--- a/arrow-array/Cargo.toml
+++ b/arrow-array/Cargo.toml
@@ -50,6 +50,9 @@ num = { version = "0.4.1", default-features = false, features = ["std"] }
 half = { version = "2.1", default-features = false, features = ["num-traits"] }
 hashbrown = { version = "0.14", default-features = false }
 
+[features]
+ffi = ["arrow-schema/ffi", "arrow-data/ffi"]
+
 [dev-dependencies]
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
 criterion = { version = "0.5", default-features = false }
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index e2839b19e..79f2d4758 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -428,6 +428,18 @@ impl BinaryViewArray {
     }
 }
 
+impl From<Vec<&[u8]>> for BinaryViewArray {
+    fn from(v: Vec<&[u8]>) -> Self {
+        Self::from_iter_values(v)
+    }
+}
+
+impl From<Vec<Option<&[u8]>>> for BinaryViewArray {
+    fn from(v: Vec<Option<&[u8]>>) -> Self {
+        v.into_iter().collect()
+    }
+}
+
 /// A [`GenericByteViewArray`] that stores utf8 data
 ///
 /// # Example
diff --git a/arrow/src/ffi.rs b/arrow-array/src/ffi.rs
similarity index 81%
rename from arrow/src/ffi.rs
rename to arrow-array/src/ffi.rs
index d33de9d65..7b988bb07 100644
--- a/arrow/src/ffi.rs
+++ b/arrow-array/src/ffi.rs
@@ -29,11 +29,11 @@
 //!
 //! ```rust
 //! # use std::sync::Arc;
-//! # use arrow::array::{Int32Array, Array, ArrayData, make_array};
-//! # use arrow::error::Result;
-//! # use arrow_arith::numeric::add;
-//! # use arrow::ffi::{to_ffi, from_ffi};
-//! # fn main() -> Result<()> {
+//! # use arrow_array::{Int32Array, Array, make_array};
+//! # use arrow_data::ArrayData;
+//! # use arrow_array::ffi::{to_ffi, from_ffi};
+//! # use arrow_schema::ArrowError;
+//! # fn main() -> Result<(), ArrowError> {
 //! // create an array natively
 //!
 //! let array = Int32Array::from(vec![Some(1), None, Some(3)]);
@@ -46,11 +46,8 @@
 //! let data = unsafe { from_ffi(out_array, &out_schema) }?;
 //! let array = Int32Array::from(data);
 //!
-//! // perform some operation
-//! let array = add(&array, &array)?;
-//!
 //! // verify
-//! assert_eq!(array.as_ref(), &Int32Array::from(vec![Some(2), None, Some(6)]));
+//! assert_eq!(array, Int32Array::from(vec![Some(1), None, Some(3)]));
 //! #
 //! # Ok(())
 //! # }
@@ -60,9 +57,9 @@
 //!
 //! ```
 //! # use std::ptr::addr_of_mut;
-//! # use arrow::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema};
+//! # use arrow_array::ffi::{from_ffi, FFI_ArrowArray};
 //! # use arrow_array::{ArrayRef, make_array};
-//! # use arrow_schema::ArrowError;
+//! # use arrow_schema::{ArrowError, ffi::FFI_ArrowSchema};
 //! #
 //! /// A foreign data container that can export to C Data interface
 //! struct ForeignArray {};
@@ -106,16 +103,39 @@ To export an array, create an `ArrowArray` using [ArrowArray::try_new].
 
 use std::{mem::size_of, ptr::NonNull, sync::Arc};
 
+use arrow_buffer::{bit_util, Buffer, MutableBuffer};
 pub use arrow_data::ffi::FFI_ArrowArray;
-pub use arrow_schema::ffi::{FFI_ArrowSchema, Flags};
+use arrow_data::{layout, ArrayData};
+pub use arrow_schema::ffi::FFI_ArrowSchema;
+use arrow_schema::{ArrowError, DataType, UnionMode};
+
+use crate::array::ArrayRef;
+
+type Result<T> = std::result::Result<T, ArrowError>;
 
-use arrow_schema::UnionMode;
+/// Exports an array to raw pointers of the C Data Interface provided by the consumer.
+/// # Safety
+/// Assumes that these pointers represent valid C Data Interfaces, both in memory
+/// representation and lifetime via the `release` mechanism.
+///
+/// This function copies the content of two FFI structs [arrow_data::ffi::FFI_ArrowArray] and
+/// [arrow_schema::ffi::FFI_ArrowSchema] in the array to the location pointed by the raw pointers.
+/// Usually the raw pointers are provided by the array data consumer.
+#[deprecated(note = "Use FFI_ArrowArray::new and FFI_ArrowSchema::try_from")]
+pub unsafe fn export_array_into_raw(
+    src: ArrayRef,
+    out_array: *mut FFI_ArrowArray,
+    out_schema: *mut FFI_ArrowSchema,
+) -> Result<()> {
+    let data = src.to_data();
+    let array = FFI_ArrowArray::new(&data);
+    let schema = FFI_ArrowSchema::try_from(data.data_type())?;
 
-use crate::array::{layout, ArrayData};
-use crate::buffer::{Buffer, MutableBuffer};
-use crate::datatypes::DataType;
-use crate::error::{ArrowError, Result};
-use crate::util::bit_util;
+    std::ptr::write_unaligned(out_array, array);
+    std::ptr::write_unaligned(out_schema, schema);
+
+    Ok(())
+}
 
 // returns the number of bits that buffer `i` (in the C data interface) is expected to have.
 // This is set by the Arrow specification
@@ -464,19 +484,17 @@ impl<'a> ImportedArrowArray<'a> {
 }
 
 #[cfg(test)]
-mod tests {
+mod tests_to_then_from_ffi {
     use std::collections::HashMap;
     use std::mem::ManuallyDrop;
-    use std::ptr::addr_of_mut;
 
-    use arrow_array::builder::UnionBuilder;
-    use arrow_array::cast::AsArray;
-    use arrow_array::types::{Float64Type, Int32Type};
-    use arrow_array::*;
     use arrow_buffer::NullBuffer;
+    use arrow_schema::Field;
 
-    use crate::compute::kernels;
-    use crate::datatypes::{Field, Int8Type};
+    use crate::builder::UnionBuilder;
+    use crate::cast::AsArray;
+    use crate::types::{Float64Type, Int32Type, Int8Type};
+    use crate::*;
 
     use super::*;
 
@@ -490,10 +508,9 @@ mod tests {
 
         // (simulate consumer) import it
         let array = Int32Array::from(unsafe { from_ffi(array, &schema) }.unwrap());
-        let array = kernels::numeric::add(&array, &array).unwrap();
 
         // verify
-        assert_eq!(array.as_ref(), &Int32Array::from(vec![2, 4, 6]));
+        assert_eq!(array, Int32Array::from(vec![1, 2, 3]));
     }
 
     #[test]
@@ -535,15 +552,9 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
         let array = array.as_any().downcast_ref::<Int32Array>().unwrap();
-        assert_eq!(array, &Int32Array::from(vec![Some(2), None]));
-
-        let array = kernels::numeric::add(array, array).unwrap();
 
-        // verify
-        assert_eq!(array.as_ref(), &Int32Array::from(vec![Some(4), None]));
+        assert_eq!(array, &Int32Array::from(vec![Some(2), None]));
 
         // (drop/release)
         Ok(())
@@ -589,21 +600,13 @@ mod tests {
         let array = make_array(data);
 
         // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array
             .as_any()
             .downcast_ref::<GenericStringArray<Offset>>()
             .unwrap();
 
         // verify
-        let expected = GenericStringArray::<Offset>::from(vec![
-            Some("a"),
-            None,
-            Some("aaa"),
-            Some("a"),
-            None,
-            Some("aaa"),
-        ]);
+        let expected = GenericStringArray::<Offset>::from(vec![Some("a"), None, Some("aaa")]);
         assert_eq!(array, &expected);
 
         // (drop/release)
@@ -694,23 +697,13 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array
             .as_any()
             .downcast_ref::<GenericBinaryArray<Offset>>()
             .unwrap();
 
         // verify
-        let expected: Vec<Option<&[u8]>> = vec![
-            Some(b"a"),
-            None,
-            Some(b"aaa"),
-            Some(b"a"),
-            None,
-            Some(b"aaa"),
-        ];
+        let expected: Vec<Option<&[u8]>> = vec![Some(b"a"), None, Some(b"aaa")];
         let expected = GenericBinaryArray::<Offset>::from(expected);
         assert_eq!(array, &expected);
 
@@ -739,15 +732,12 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
         let array = array.as_any().downcast_ref::<BooleanArray>().unwrap();
-        let array = kernels::boolean::not(array)?;
 
         // verify
         assert_eq!(
             array,
-            BooleanArray::from(vec![None, Some(false), Some(true)])
+            &BooleanArray::from(vec![None, Some(true), Some(false)])
         );
 
         // (drop/release)
@@ -765,9 +755,6 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array
             .as_any()
             .downcast_ref::<Time32MillisecondArray>()
@@ -776,7 +763,7 @@ mod tests {
         // verify
         assert_eq!(
             array,
-            &Time32MillisecondArray::from(vec![None, Some(1), Some(2), None, Some(1), Some(2)])
+            &Time32MillisecondArray::from(vec![None, Some(1), Some(2)])
         );
 
         // (drop/release)
@@ -794,9 +781,6 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array
             .as_any()
             .downcast_ref::<TimestampMillisecondArray>()
@@ -805,7 +789,7 @@ mod tests {
         // verify
         assert_eq!(
             array,
-            &TimestampMillisecondArray::from(vec![None, Some(1), Some(2), None, Some(1), Some(2)])
+            &TimestampMillisecondArray::from(vec![None, Some(1), Some(2)])
         );
 
         // (drop/release)
@@ -830,9 +814,6 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array
             .as_any()
             .downcast_ref::<FixedSizeBinaryArray>()
@@ -849,12 +830,6 @@ mod tests {
                     Some(vec![20, 20, 20]),
                     Some(vec![30, 30, 30]),
                     None,
-                    None,
-                    Some(vec![10, 10, 10]),
-                    None,
-                    Some(vec![20, 20, 20]),
-                    Some(vec![30, 30, 30]),
-                    None,
                 ]
                 .into_iter(),
                 3
@@ -891,9 +866,6 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array.as_any().downcast_ref::<FixedSizeListArray>().unwrap();
 
         // 0010 0100
@@ -903,15 +875,14 @@ mod tests {
 
         let mut w = vec![];
         w.extend_from_slice(&v);
-        w.extend_from_slice(&v);
 
         let expected_value_data = ArrayData::builder(DataType::Int32)
-            .len(18)
+            .len(9)
             .add_buffer(Buffer::from_slice_ref(&w))
             .build()?;
 
         let expected_list_data = ArrayData::builder(list_data_type)
-            .len(6)
+            .len(3)
             .null_bit_buffer(Some(Buffer::from(expected_validity_bits)))
             .add_child_data(expected_value_data)
             .build()?;
@@ -936,16 +907,13 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let actual = array
             .as_any()
             .downcast_ref::<DictionaryArray<Int8Type>>()
             .unwrap();
 
         // verify
-        let new_values = vec!["a", "aaa", "aaa", "a", "aaa", "aaa"];
+        let new_values = vec!["a", "aaa", "aaa"];
         let expected: DictionaryArray<Int8Type> = new_values.into_iter().collect();
         assert_eq!(actual, &expected);
 
@@ -956,7 +924,6 @@ mod tests {
     #[test]
     #[allow(deprecated)]
     fn test_export_array_into_raw() -> Result<()> {
-        use crate::array::export_array_into_raw;
         let array = make_array(Int32Array::from(vec![1, 2, 3]).into_data());
 
         // Assume two raw pointers provided by the consumer
@@ -964,8 +931,8 @@ mod tests {
         let mut out_schema = FFI_ArrowSchema::empty();
 
         {
-            let out_array_ptr = addr_of_mut!(out_array);
-            let out_schema_ptr = addr_of_mut!(out_schema);
+            let out_array_ptr = std::ptr::addr_of_mut!(out_array);
+            let out_schema_ptr = std::ptr::addr_of_mut!(out_schema);
             unsafe {
                 export_array_into_raw(array, out_array_ptr, out_schema_ptr)?;
             }
@@ -977,10 +944,9 @@ mod tests {
 
         // perform some operation
         let array = array.as_any().downcast_ref::<Int32Array>().unwrap();
-        let array = kernels::numeric::add(array, array).unwrap();
 
         // verify
-        assert_eq!(array.as_ref(), &Int32Array::from(vec![2, 4, 6]));
+        assert_eq!(array, &Int32Array::from(vec![1, 2, 3]));
         Ok(())
     }
 
@@ -995,9 +961,6 @@ mod tests {
         // (simulate consumer) import it
         let data = unsafe { from_ffi(array, &schema) }?;
         let array = make_array(data);
-
-        // perform some operation
-        let array = kernels::concat::concat(&[array.as_ref(), array.as_ref()]).unwrap();
         let array = array
             .as_any()
             .downcast_ref::<DurationSecondArray>()
@@ -1006,7 +969,7 @@ mod tests {
         // verify
         assert_eq!(
             array,
-            &DurationSecondArray::from(vec![None, Some(1), Some(2), None, Some(1), Some(2)])
+            &DurationSecondArray::from(vec![None, Some(1), Some(2)])
         );
 
         // (drop/release)
@@ -1248,3 +1211,213 @@ mod tests {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests_from_ffi {
+    use std::sync::Arc;
+
+    use arrow_buffer::{bit_util, buffer::Buffer};
+    use arrow_data::ArrayData;
+    use arrow_schema::{DataType, Field};
+
+    use crate::{
+        array::{
+            Array, BooleanArray, DictionaryArray, FixedSizeBinaryArray, FixedSizeListArray,
+            Int32Array, Int64Array, StringArray, StructArray, UInt32Array, UInt64Array,
+        },
+        ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema},
+    };
+
+    use super::Result;
+
+    fn test_round_trip(expected: &ArrayData) -> Result<()> {
+        // here we export the array
+        let array = FFI_ArrowArray::new(expected);
+        let schema = FFI_ArrowSchema::try_from(expected.data_type())?;
+
+        // simulate an external consumer by being the consumer
+        let result = &unsafe { from_ffi(array, &schema) }?;
+
+        assert_eq!(result, expected);
+        Ok(())
+    }
+
+    #[test]
+    fn test_u32() -> Result<()> {
+        let array = UInt32Array::from(vec![Some(2), None, Some(1), None]);
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_u64() -> Result<()> {
+        let array = UInt64Array::from(vec![Some(2), None, Some(1), None]);
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_i64() -> Result<()> {
+        let array = Int64Array::from(vec![Some(2), None, Some(1), None]);
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_struct() -> Result<()> {
+        let inner = StructArray::from(vec![
+            (
+                Arc::new(Field::new("a1", DataType::Boolean, false)),
+                Arc::new(BooleanArray::from(vec![true, true, false, false])) as Arc<dyn Array>,
+            ),
+            (
+                Arc::new(Field::new("a2", DataType::UInt32, false)),
+                Arc::new(UInt32Array::from(vec![1, 2, 3, 4])),
+            ),
+        ]);
+
+        let array = StructArray::from(vec![
+            (
+                Arc::new(Field::new("a", inner.data_type().clone(), false)),
+                Arc::new(inner) as Arc<dyn Array>,
+            ),
+            (
+                Arc::new(Field::new("b", DataType::Boolean, false)),
+                Arc::new(BooleanArray::from(vec![false, false, true, true])) as Arc<dyn Array>,
+            ),
+            (
+                Arc::new(Field::new("c", DataType::UInt32, false)),
+                Arc::new(UInt32Array::from(vec![42, 28, 19, 31])),
+            ),
+        ]);
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_dictionary() -> Result<()> {
+        let values = StringArray::from(vec![Some("foo"), Some("bar"), None]);
+        let keys = Int32Array::from(vec![
+            Some(0),
+            Some(1),
+            None,
+            Some(1),
+            Some(1),
+            None,
+            Some(1),
+            Some(2),
+            Some(1),
+            None,
+        ]);
+        let array = DictionaryArray::new(keys, Arc::new(values));
+
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_fixed_size_binary() -> Result<()> {
+        let values = vec![vec![10, 10, 10], vec![20, 20, 20], vec![30, 30, 30]];
+        let array = FixedSizeBinaryArray::try_from_iter(values.into_iter())?;
+
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_fixed_size_binary_with_nulls() -> Result<()> {
+        let values = vec![
+            None,
+            Some(vec![10, 10, 10]),
+            None,
+            Some(vec![20, 20, 20]),
+            Some(vec![30, 30, 30]),
+            None,
+        ];
+        let array = FixedSizeBinaryArray::try_from_sparse_iter_with_size(values.into_iter(), 3)?;
+
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_fixed_size_list() -> Result<()> {
+        let v: Vec<i64> = (0..9).collect();
+        let value_data = ArrayData::builder(DataType::Int64)
+            .len(9)
+            .add_buffer(Buffer::from_slice_ref(v))
+            .build()?;
+        let list_data_type =
+            DataType::FixedSizeList(Arc::new(Field::new("f", DataType::Int64, false)), 3);
+        let list_data = ArrayData::builder(list_data_type)
+            .len(3)
+            .add_child_data(value_data)
+            .build()?;
+        let array = FixedSizeListArray::from(list_data);
+
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_fixed_size_list_with_nulls() -> Result<()> {
+        // 0100 0110
+        let mut validity_bits: [u8; 1] = [0; 1];
+        bit_util::set_bit(&mut validity_bits, 1);
+        bit_util::set_bit(&mut validity_bits, 2);
+        bit_util::set_bit(&mut validity_bits, 6);
+
+        let v: Vec<i16> = (0..16).collect();
+        let value_data = ArrayData::builder(DataType::Int16)
+            .len(16)
+            .add_buffer(Buffer::from_slice_ref(v))
+            .build()?;
+        let list_data_type =
+            DataType::FixedSizeList(Arc::new(Field::new("f", DataType::Int16, false)), 2);
+        let list_data = ArrayData::builder(list_data_type)
+            .len(8)
+            .null_bit_buffer(Some(Buffer::from(validity_bits)))
+            .add_child_data(value_data)
+            .build()?;
+        let array = FixedSizeListArray::from(list_data);
+
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+
+    #[test]
+    fn test_fixed_size_list_nested() -> Result<()> {
+        let v: Vec<i32> = (0..16).collect();
+        let value_data = ArrayData::builder(DataType::Int32)
+            .len(16)
+            .add_buffer(Buffer::from_slice_ref(v))
+            .build()?;
+
+        let offsets: Vec<i32> = vec![0, 2, 4, 6, 8, 10, 12, 14, 16];
+        let value_offsets = Buffer::from_slice_ref(offsets);
+        let inner_list_data_type =
+            DataType::List(Arc::new(Field::new("item", DataType::Int32, false)));
+        let inner_list_data = ArrayData::builder(inner_list_data_type.clone())
+            .len(8)
+            .add_buffer(value_offsets)
+            .add_child_data(value_data)
+            .build()?;
+
+        // 0000 0100
+        let mut validity_bits: [u8; 1] = [0; 1];
+        bit_util::set_bit(&mut validity_bits, 2);
+
+        let list_data_type =
+            DataType::FixedSizeList(Arc::new(Field::new("f", inner_list_data_type, false)), 2);
+        let list_data = ArrayData::builder(list_data_type)
+            .len(4)
+            .null_bit_buffer(Some(Buffer::from(validity_bits)))
+            .add_child_data(inner_list_data)
+            .build()?;
+
+        let array = FixedSizeListArray::from(list_data);
+
+        let data = array.into_data();
+        test_round_trip(&data)
+    }
+}
diff --git a/arrow/src/ffi_stream.rs b/arrow-array/src/ffi_stream.rs
similarity index 98%
rename from arrow/src/ffi_stream.rs
rename to arrow-array/src/ffi_stream.rs
index 15b88ef32..6f3405ead 100644
--- a/arrow/src/ffi_stream.rs
+++ b/arrow-array/src/ffi_stream.rs
@@ -63,14 +63,16 @@ use std::{
     sync::Arc,
 };
 
+use arrow_data::ffi::FFI_ArrowArray;
+use arrow_schema::{ffi::FFI_ArrowSchema, ArrowError, Schema, SchemaRef};
+
 use crate::array::Array;
 use crate::array::StructArray;
-use crate::datatypes::{Schema, SchemaRef};
-use crate::error::ArrowError;
-use crate::error::Result;
-use crate::ffi::*;
+use crate::ffi::from_ffi_and_data_type;
 use crate::record_batch::{RecordBatch, RecordBatchReader};
 
+type Result<T> = std::result::Result<T, ArrowError>;
+
 const ENOMEM: i32 = 12;
 const EIO: i32 = 5;
 const EINVAL: i32 = 22;
@@ -81,6 +83,7 @@ const ENOSYS: i32 = 78;
 /// This was created by bindgen
 #[repr(C)]
 #[derive(Debug)]
+#[allow(missing_docs)]
 pub struct FFI_ArrowArrayStream {
     pub get_schema: Option<
         unsafe extern "C" fn(arg1: *mut FFI_ArrowArrayStream, out: *mut FFI_ArrowSchema) -> c_int,
@@ -393,8 +396,10 @@ pub unsafe fn export_reader_into_raw(
 mod tests {
     use super::*;
 
+    use arrow_schema::Field;
+
     use crate::array::Int32Array;
-    use crate::datatypes::Field;
+    use crate::ffi::from_ffi;
 
     struct TestRecordBatchReader {
         schema: SchemaRef,
diff --git a/arrow-array/src/lib.rs b/arrow-array/src/lib.rs
index ef98c5efe..90bc5e312 100644
--- a/arrow-array/src/lib.rs
+++ b/arrow-array/src/lib.rs
@@ -197,6 +197,10 @@ pub use scalar::*;
 pub mod builder;
 pub mod cast;
 mod delta;
+#[cfg(feature = "ffi")]
+pub mod ffi;
+#[cfg(feature = "ffi")]
+pub mod ffi_stream;
 pub mod iterator;
 pub mod run_iterator;
 pub mod temporal_conversions;
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index 8b7579c4c..36072760e 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -46,6 +46,8 @@ use crate::cast::dictionary::*;
 use crate::cast::list::*;
 use crate::cast::string::*;
 
+use arrow_buffer::ScalarBuffer;
+use arrow_data::ByteView;
 use chrono::{NaiveTime, Offset, TimeZone, Utc};
 use std::cmp::Ordering;
 use std::sync::Arc;
@@ -119,6 +121,8 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
             | Utf8
             | LargeBinary
             | LargeUtf8
+            | BinaryView
+            | Utf8View
             | List(_)
             | LargeList(_)
             | FixedSizeList(_, _)
@@ -192,8 +196,8 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
             DataType::is_integer(to_type) || DataType::is_floating(to_type) || to_type == &Utf8 || to_type == &LargeUtf8
         }
 
-        (Binary, LargeBinary | Utf8 | LargeUtf8 | FixedSizeBinary(_)) => true,
-        (LargeBinary, Binary | Utf8 | LargeUtf8 | FixedSizeBinary(_)) => true,
+        (Binary, LargeBinary | Utf8 | LargeUtf8 | FixedSizeBinary(_) | BinaryView) => true,
+        (LargeBinary, Binary | Utf8 | LargeUtf8 | FixedSizeBinary(_) | BinaryView) => true,
         (FixedSizeBinary(_), Binary | LargeBinary) => true,
         (
             Utf8 | LargeUtf8,
@@ -213,6 +217,7 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
             | Timestamp(Nanosecond, _)
             | Interval(_),
         ) => true,
+        (Utf8 | LargeUtf8, Utf8View) => true,
         (Utf8 | LargeUtf8, _) => to_type.is_numeric() && to_type != &Float16,
         (_, Utf8 | LargeUtf8) => from_type.is_primitive(),
 
@@ -611,6 +616,8 @@ pub fn cast_with_options(
             | Utf8
             | LargeBinary
             | LargeUtf8
+            | BinaryView
+            | Utf8View
             | List(_)
             | LargeList(_)
             | FixedSizeList(_, _)
@@ -1120,6 +1127,7 @@ pub fn cast_with_options(
                 let binary = BinaryArray::from(array.as_string::<i32>().clone());
                 cast_byte_container::<BinaryType, LargeBinaryType>(&binary)
             }
+            Utf8View => cast_byte_to_view::<Utf8Type, StringViewType>(array),
             LargeUtf8 => cast_byte_container::<Utf8Type, LargeUtf8Type>(array),
             Time32(TimeUnit::Second) => parse_string::<Time32SecondType, i32>(array, cast_options),
             Time32(TimeUnit::Millisecond) => {
@@ -1179,6 +1187,7 @@ pub fn cast_with_options(
             LargeBinary => Ok(Arc::new(LargeBinaryArray::from(
                 array.as_string::<i64>().clone(),
             ))),
+            Utf8View => cast_byte_to_view::<LargeUtf8Type, StringViewType>(array),
             Time32(TimeUnit::Second) => parse_string::<Time32SecondType, i64>(array, cast_options),
             Time32(TimeUnit::Millisecond) => {
                 parse_string::<Time32MillisecondType, i64>(array, cast_options)
@@ -1226,6 +1235,7 @@ pub fn cast_with_options(
             FixedSizeBinary(size) => {
                 cast_binary_to_fixed_size_binary::<i32>(array, *size, cast_options)
             }
+            BinaryView => cast_byte_to_view::<BinaryType, BinaryViewType>(array),
             _ => Err(ArrowError::CastError(format!(
                 "Casting from {from_type:?} to {to_type:?} not supported",
             ))),
@@ -1240,6 +1250,7 @@ pub fn cast_with_options(
             FixedSizeBinary(size) => {
                 cast_binary_to_fixed_size_binary::<i64>(array, *size, cast_options)
             }
+            BinaryView => cast_byte_to_view::<LargeBinaryType, BinaryViewType>(array),
             _ => Err(ArrowError::CastError(format!(
                 "Casting from {from_type:?} to {to_type:?} not supported",
             ))),
@@ -2238,6 +2249,56 @@ where
     Ok(Arc::new(GenericByteArray::<TO>::from(array_data)))
 }
 
+/// Helper function to cast from one `ByteArrayType` array to `ByteViewType` array.
+fn cast_byte_to_view<FROM, V>(array: &dyn Array) -> Result<ArrayRef, ArrowError>
+where
+    FROM: ByteArrayType,
+    FROM::Offset: OffsetSizeTrait + ToPrimitive,
+    V: ByteViewType,
+{
+    let data = array.to_data();
+    assert_eq!(data.data_type(), &FROM::DATA_TYPE);
+
+    let len = array.len();
+    let str_values_buf = data.buffers()[1].clone();
+    let offsets = data.buffers()[0].typed_data::<FROM::Offset>();
+
+    let mut views_builder = BufferBuilder::<u128>::new(len);
+    for w in offsets.windows(2) {
+        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));
+        } 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());
+        }
+    }
+
+    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(),
+        )
+    }))
+}
+
 #[cfg(test)]
 mod tests {
     use arrow_buffer::{Buffer, NullBuffer};
@@ -5044,6 +5105,70 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_string_to_view() {
+        _test_string_to_view::<i32>();
+        _test_string_to_view::<i64>();
+    }
+
+    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());
+
+        assert!(can_cast_types(
+            string_array.data_type(),
+            &DataType::Utf8View
+        ));
+
+        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);
+        assert_eq!(string_view_array.as_ref(), &expect_string_view_array);
+    }
+
+    #[test]
+    fn test_bianry_to_view() {
+        _test_binary_to_view::<i32>();
+        _test_binary_to_view::<i64>();
+    }
+
+    fn _test_binary_to_view<O>()
+    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());
+
+        assert!(can_cast_types(
+            binary_array.data_type(),
+            &DataType::BinaryView
+        ));
+
+        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);
+        assert_eq!(binary_view_array.as_ref(), &expect_binary_view_array);
+    }
+
     #[test]
     fn test_cast_from_f64() {
         let f64_values: Vec<f64> = vec![
diff --git a/arrow-data/src/data.rs b/arrow-data/src/data.rs
index 0ea74f789..d092fd049 100644
--- a/arrow-data/src/data.rs
+++ b/arrow-data/src/data.rs
@@ -22,9 +22,9 @@ use crate::bit_iterator::BitSliceIterator;
 use arrow_buffer::buffer::{BooleanBuffer, NullBuffer};
 use arrow_buffer::{bit_util, i256, ArrowNativeType, Buffer, 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};
 
@@ -929,6 +929,41 @@ impl ArrayData {
         Ok(())
     }
 
+    /// Does a cheap sanity check that the `self.len` values in `buffer` are valid
+    /// offsets and sizes (of type T) into some other buffer of `values_length` bytes long
+    fn validate_offsets_and_sizes<T: ArrowNativeType + num::Num + std::fmt::Display>(
+        &self,
+        values_length: usize,
+    ) -> Result<(), ArrowError> {
+        let offsets: &[T] = self.typed_buffer(0, self.len)?;
+        let sizes: &[T] = self.typed_buffer(1, self.len)?;
+        for i in 0..values_length {
+            let size = sizes[i].to_usize().ok_or_else(|| {
+                ArrowError::InvalidArgumentError(format!(
+                    "Error converting size[{}] ({}) to usize for {}",
+                    i, sizes[i], self.data_type
+                ))
+            })?;
+            let offset = offsets[i].to_usize().ok_or_else(|| {
+                ArrowError::InvalidArgumentError(format!(
+                    "Error converting offset[{}] ({}) to usize for {}",
+                    i, offsets[i], self.data_type
+                ))
+            })?;
+            if size
+                .checked_add(offset)
+                .expect("Offset and size have exceeded the usize boundary")
+                > values_length
+            {
+                return Err(ArrowError::InvalidArgumentError(format!(
+                    "Size {} at index {} is larger than the remaining values for {}",
+                    size, i, self.data_type
+                )));
+            }
+        }
+        Ok(())
+    }
+
     /// Validates the layout of `child_data` ArrayData structures
     fn validate_child_data(&self) -> Result<(), ArrowError> {
         match &self.data_type {
@@ -942,6 +977,16 @@ impl ArrayData {
                 self.validate_offsets::<i64>(values_data.len)?;
                 Ok(())
             }
+            DataType::ListView(field) => {
+                let values_data = self.get_single_valid_child_data(field.data_type())?;
+                self.validate_offsets_and_sizes::<i32>(values_data.len)?;
+                Ok(())
+            }
+            DataType::LargeListView(field) => {
+                let values_data = self.get_single_valid_child_data(field.data_type())?;
+                self.validate_offsets_and_sizes::<i64>(values_data.len)?;
+                Ok(())
+            }
             DataType::FixedSizeList(field, list_size) => {
                 let values_data = self.get_single_valid_child_data(field.data_type())?;
 
@@ -1546,9 +1591,8 @@ pub fn layout(data_type: &DataType) -> DataTypeLayout {
         DataType::BinaryView | DataType::Utf8View => DataTypeLayout::new_view(),
         DataType::FixedSizeList(_, _) => DataTypeLayout::new_nullable_empty(), // all in child data
         DataType::List(_) => DataTypeLayout::new_fixed_width::<i32>(),
-        DataType::ListView(_) | DataType::LargeListView(_) => {
-            unimplemented!("ListView/LargeListView not implemented")
-        }
+        DataType::ListView(_) => DataTypeLayout::new_list_view::<i32>(),
+        DataType::LargeListView(_) => DataTypeLayout::new_list_view::<i64>(),
         DataType::LargeList(_) => DataTypeLayout::new_fixed_width::<i64>(),
         DataType::Map(_, _) => DataTypeLayout::new_fixed_width::<i32>(),
         DataType::Struct(_) => DataTypeLayout::new_nullable_empty(), // all in child data,
@@ -1661,6 +1705,24 @@ impl DataTypeLayout {
             variadic: true,
         }
     }
+
+    /// Describes a list view type
+    pub fn new_list_view<T>() -> Self {
+        Self {
+            buffers: vec![
+                BufferSpec::FixedWidth {
+                    byte_width: mem::size_of::<T>(),
+                    alignment: mem::align_of::<T>(),
+                },
+                BufferSpec::FixedWidth {
+                    byte_width: mem::size_of::<T>(),
+                    alignment: mem::align_of::<T>(),
+                },
+            ],
+            can_contain_null_mask: true,
+            variadic: true,
+        }
+    }
 }
 
 /// Layout specification for a single data type buffer
diff --git a/arrow-flight/examples/flight_sql_server.rs b/arrow-flight/examples/flight_sql_server.rs
index a8f8d1606..031628eaa 100644
--- a/arrow-flight/examples/flight_sql_server.rs
+++ b/arrow-flight/examples/flight_sql_server.rs
@@ -24,7 +24,9 @@ use once_cell::sync::Lazy;
 use prost::Message;
 use std::collections::HashSet;
 use std::pin::Pin;
+use std::str::FromStr;
 use std::sync::Arc;
+use tonic::metadata::MetadataValue;
 use tonic::transport::Server;
 use tonic::transport::{Certificate, Identity, ServerTlsConfig};
 use tonic::{Request, Response, Status, Streaming};
@@ -52,7 +54,7 @@ use arrow_flight::utils::batches_to_flight_data;
 use arrow_flight::{
     flight_service_server::FlightService, flight_service_server::FlightServiceServer, Action,
     FlightData, FlightDescriptor, FlightEndpoint, FlightInfo, HandshakeRequest, HandshakeResponse,
-    IpcMessage, Location, SchemaAsIpc, Ticket,
+    IpcMessage, SchemaAsIpc, Ticket,
 };
 use arrow_ipc::writer::IpcWriteOptions;
 use arrow_schema::{ArrowError, DataType, Field, Schema};
@@ -184,7 +186,15 @@ impl FlightSqlService for FlightSqlServiceImpl {
         };
         let result = Ok(result);
         let output = futures::stream::iter(vec![result]);
-        return Ok(Response::new(Box::pin(output)));
+
+        let token = format!("Bearer {}", FAKE_TOKEN);
+        let mut response: Response<Pin<Box<dyn Stream<Item = _> + Send>>> =
+            Response::new(Box::pin(output));
+        response.metadata_mut().append(
+            "authorization",
+            MetadataValue::from_str(token.as_str()).unwrap(),
+        );
+        return Ok(response);
     }
 
     async fn do_get_fallback(
@@ -235,13 +245,12 @@ impl FlightSqlService for FlightSqlServiceImpl {
         self.check_token(&request)?;
         let handle = std::str::from_utf8(&cmd.prepared_statement_handle)
             .map_err(|e| status!("Unable to parse handle", e))?;
+
         let batch = Self::fake_result().map_err(|e| status!("Could not fake a result", e))?;
         let schema = (*batch.schema()).clone();
         let num_rows = batch.num_rows();
         let num_bytes = batch.get_array_memory_size();
-        let loc = Location {
-            uri: "grpc+tcp://127.0.0.1".to_string(),
-        };
+
         let fetch = FetchResults {
             handle: handle.to_string(),
         };
@@ -249,7 +258,7 @@ impl FlightSqlService for FlightSqlServiceImpl {
         let ticket = Ticket { ticket: buf };
         let endpoint = FlightEndpoint {
             ticket: Some(ticket),
-            location: vec![loc],
+            location: vec![],
             expiration_time: None,
             app_metadata: vec![].into(),
         };
@@ -662,9 +671,7 @@ impl FlightSqlService for FlightSqlServiceImpl {
         _query: ActionClosePreparedStatementRequest,
         _request: Request<Action>,
     ) -> Result<(), Status> {
-        Err(Status::unimplemented(
-            "Implement do_action_close_prepared_statement",
-        ))
+        Ok(())
     }
 
     async fn do_action_create_prepared_substrait_plan(
@@ -725,9 +732,8 @@ impl FlightSqlService for FlightSqlServiceImpl {
 /// This example shows how to run a FlightSql server
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let addr = "0.0.0.0:50051".parse()?;
-
-    let svc = FlightServiceServer::new(FlightSqlServiceImpl {});
+    let addr_str = "0.0.0.0:50051";
+    let addr = addr_str.parse()?;
 
     println!("Listening on {:?}", addr);
 
@@ -736,6 +742,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let key = std::fs::read_to_string("arrow-flight/examples/data/server.key")?;
         let client_ca = std::fs::read_to_string("arrow-flight/examples/data/client_ca.pem")?;
 
+        let svc = FlightServiceServer::new(FlightSqlServiceImpl {});
         let tls_config = ServerTlsConfig::new()
             .identity(Identity::from_pem(&cert, &key))
             .client_ca_root(Certificate::from_pem(&client_ca));
@@ -746,6 +753,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .serve(addr)
             .await?;
     } else {
+        let svc = FlightServiceServer::new(FlightSqlServiceImpl {});
+
         Server::builder().add_service(svc).serve(addr).await?;
     }
 
@@ -999,15 +1008,6 @@ mod tests {
                 .to_string()
                 .contains("Invalid credentials"));
 
-            // forget to set_token
-            client.handshake("admin", "password").await.unwrap();
-            assert!(client
-                .prepare("select 1;".to_string(), None)
-                .await
-                .unwrap_err()
-                .to_string()
-                .contains("No authorization header"));
-
             // Invalid Tokens
             client.handshake("admin", "password").await.unwrap();
             client.set_token("wrong token".to_string());
@@ -1017,6 +1017,12 @@ mod tests {
                 .unwrap_err()
                 .to_string()
                 .contains("invalid token"));
+
+            client.clear_token();
+
+            // Successful call (token is automatically set by handshake)
+            client.handshake("admin", "password").await.unwrap();
+            client.prepare("select 1;".to_string(), None).await.unwrap();
         })
         .await
     }
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index 44250fbe6..29782a2bc 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -97,6 +97,11 @@ impl FlightSqlServiceClient<Channel> {
         self.token = Some(token);
     }
 
+    /// Clear the auth token.
+    pub fn clear_token(&mut self) {
+        self.token = None;
+    }
+
     /// Set header value.
     pub fn set_header(&mut self, key: impl Into<String>, value: impl Into<String>) {
         let key: String = key.into();
diff --git a/arrow-json/src/lib.rs b/arrow-json/src/lib.rs
index c7839a171..b6c441012 100644
--- a/arrow-json/src/lib.rs
+++ b/arrow-json/src/lib.rs
@@ -69,18 +69,6 @@
 pub mod reader;
 pub mod writer;
 
-#[doc(hidden)]
-#[deprecated(note = "Use Decoder")]
-pub type RawDecoder = reader::Decoder;
-
-#[doc(hidden)]
-#[deprecated(note = "Use Reader")]
-pub type RawReader<R> = Reader<R>;
-
-#[doc(hidden)]
-#[deprecated(note = "Use ReaderBuilder")]
-pub type RawReaderBuilder = ReaderBuilder;
-
 pub use self::reader::{Reader, ReaderBuilder};
 pub use self::writer::{ArrayWriter, LineDelimitedWriter, Writer, WriterBuilder};
 use half::f16;
diff --git a/arrow-json/src/writer.rs b/arrow-json/src/writer.rs
index b97065989..c9619f80b 100644
--- a/arrow-json/src/writer.rs
+++ b/arrow-json/src/writer.rs
@@ -106,427 +106,14 @@
 //! ```
 mod encoder;
 
-use std::iter;
 use std::{fmt::Debug, io::Write};
 
-use serde_json::map::Map as JsonMap;
-use serde_json::Value;
-
-use crate::JsonSerializable;
-use arrow_array::cast::*;
-use arrow_array::types::*;
 use arrow_array::*;
 use arrow_schema::*;
 
 use crate::writer::encoder::EncoderOptions;
-use arrow_cast::display::{ArrayFormatter, FormatOptions};
 use encoder::make_encoder;
 
-fn primitive_array_to_json<T>(array: &dyn Array) -> Result<Vec<Value>, ArrowError>
-where
-    T: ArrowPrimitiveType,
-    T::Native: JsonSerializable,
-{
-    Ok(array
-        .as_primitive::<T>()
-        .iter()
-        .map(|maybe_value| match maybe_value {
-            Some(v) => v.into_json_value().unwrap_or(Value::Null),
-            None => Value::Null,
-        })
-        .collect())
-}
-
-fn struct_array_to_jsonmap_array(
-    array: &StructArray,
-    explicit_nulls: bool,
-) -> Result<Vec<Option<JsonMap<String, Value>>>, ArrowError> {
-    let inner_col_names = array.column_names();
-
-    let mut inner_objs = (0..array.len())
-        // Ensure we write nulls for struct arrays as nulls in JSON
-        // Instead of writing a struct with nulls
-        .map(|index| array.is_valid(index).then(JsonMap::new))
-        .collect::<Vec<Option<JsonMap<String, Value>>>>();
-
-    for (j, struct_col) in array.columns().iter().enumerate() {
-        set_column_for_json_rows(
-            &mut inner_objs,
-            struct_col,
-            inner_col_names[j],
-            explicit_nulls,
-        )?
-    }
-    Ok(inner_objs)
-}
-
-/// Converts an arrow [`Array`] into a `Vec` of Serde JSON [`serde_json::Value`]'s
-#[deprecated(note = "Use Writer")]
-pub fn array_to_json_array(array: &dyn Array) -> Result<Vec<Value>, ArrowError> {
-    // For backwards compatibility, default to skip nulls
-    array_to_json_array_internal(array, false)
-}
-
-fn array_to_json_array_internal(
-    array: &dyn Array,
-    explicit_nulls: bool,
-) -> Result<Vec<Value>, ArrowError> {
-    match array.data_type() {
-        DataType::Null => Ok(iter::repeat(Value::Null).take(array.len()).collect()),
-        DataType::Boolean => Ok(array
-            .as_boolean()
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => v.into(),
-                None => Value::Null,
-            })
-            .collect()),
-
-        DataType::Utf8 => Ok(array
-            .as_string::<i32>()
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => v.into(),
-                None => Value::Null,
-            })
-            .collect()),
-        DataType::LargeUtf8 => Ok(array
-            .as_string::<i64>()
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => v.into(),
-                None => Value::Null,
-            })
-            .collect()),
-        DataType::Int8 => primitive_array_to_json::<Int8Type>(array),
-        DataType::Int16 => primitive_array_to_json::<Int16Type>(array),
-        DataType::Int32 => primitive_array_to_json::<Int32Type>(array),
-        DataType::Int64 => primitive_array_to_json::<Int64Type>(array),
-        DataType::UInt8 => primitive_array_to_json::<UInt8Type>(array),
-        DataType::UInt16 => primitive_array_to_json::<UInt16Type>(array),
-        DataType::UInt32 => primitive_array_to_json::<UInt32Type>(array),
-        DataType::UInt64 => primitive_array_to_json::<UInt64Type>(array),
-        DataType::Float16 => primitive_array_to_json::<Float16Type>(array),
-        DataType::Float32 => primitive_array_to_json::<Float32Type>(array),
-        DataType::Float64 => primitive_array_to_json::<Float64Type>(array),
-        DataType::List(_) => as_list_array(array)
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => Ok(Value::Array(array_to_json_array_internal(
-                    &v,
-                    explicit_nulls,
-                )?)),
-                None => Ok(Value::Null),
-            })
-            .collect(),
-        DataType::LargeList(_) => as_large_list_array(array)
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => Ok(Value::Array(array_to_json_array_internal(
-                    &v,
-                    explicit_nulls,
-                )?)),
-                None => Ok(Value::Null),
-            })
-            .collect(),
-        DataType::FixedSizeList(_, _) => as_fixed_size_list_array(array)
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => Ok(Value::Array(array_to_json_array_internal(
-                    &v,
-                    explicit_nulls,
-                )?)),
-                None => Ok(Value::Null),
-            })
-            .collect(),
-        DataType::Struct(_) => {
-            let jsonmaps = struct_array_to_jsonmap_array(array.as_struct(), explicit_nulls)?;
-            let json_values = jsonmaps
-                .into_iter()
-                .map(|maybe_map| maybe_map.map(Value::Object).unwrap_or(Value::Null))
-                .collect();
-            Ok(json_values)
-        }
-        DataType::Map(_, _) => as_map_array(array)
-            .iter()
-            .map(|maybe_value| match maybe_value {
-                Some(v) => Ok(Value::Array(array_to_json_array_internal(
-                    &v,
-                    explicit_nulls,
-                )?)),
-                None => Ok(Value::Null),
-            })
-            .collect(),
-        t => Err(ArrowError::JsonError(format!(
-            "data type {t:?} not supported"
-        ))),
-    }
-}
-
-macro_rules! set_column_by_array_type {
-    ($cast_fn:ident, $col_name:ident, $rows:ident, $array:ident, $explicit_nulls:ident) => {
-        let arr = $cast_fn($array);
-        $rows
-            .iter_mut()
-            .zip(arr.iter())
-            .filter_map(|(maybe_row, maybe_value)| maybe_row.as_mut().map(|row| (row, maybe_value)))
-            .for_each(|(row, maybe_value)| {
-                if let Some(j) = maybe_value.map(Into::into) {
-                    row.insert($col_name.to_string(), j);
-                } else if $explicit_nulls {
-                    row.insert($col_name.to_string(), Value::Null);
-                }
-            });
-    };
-}
-
-fn set_column_by_primitive_type<T>(
-    rows: &mut [Option<JsonMap<String, Value>>],
-    array: &ArrayRef,
-    col_name: &str,
-    explicit_nulls: bool,
-) where
-    T: ArrowPrimitiveType,
-    T::Native: JsonSerializable,
-{
-    let primitive_arr = array.as_primitive::<T>();
-
-    rows.iter_mut()
-        .zip(primitive_arr.iter())
-        .filter_map(|(maybe_row, maybe_value)| maybe_row.as_mut().map(|row| (row, maybe_value)))
-        .for_each(|(row, maybe_value)| {
-            if let Some(j) = maybe_value.and_then(|v| v.into_json_value()) {
-                row.insert(col_name.to_string(), j);
-            } else if explicit_nulls {
-                row.insert(col_name.to_string(), Value::Null);
-            }
-        });
-}
-
-fn set_column_for_json_rows(
-    rows: &mut [Option<JsonMap<String, Value>>],
-    array: &ArrayRef,
-    col_name: &str,
-    explicit_nulls: bool,
-) -> Result<(), ArrowError> {
-    match array.data_type() {
-        DataType::Int8 => {
-            set_column_by_primitive_type::<Int8Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Int16 => {
-            set_column_by_primitive_type::<Int16Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Int32 => {
-            set_column_by_primitive_type::<Int32Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Int64 => {
-            set_column_by_primitive_type::<Int64Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::UInt8 => {
-            set_column_by_primitive_type::<UInt8Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::UInt16 => {
-            set_column_by_primitive_type::<UInt16Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::UInt32 => {
-            set_column_by_primitive_type::<UInt32Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::UInt64 => {
-            set_column_by_primitive_type::<UInt64Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Float16 => {
-            set_column_by_primitive_type::<Float16Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Float32 => {
-            set_column_by_primitive_type::<Float32Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Float64 => {
-            set_column_by_primitive_type::<Float64Type>(rows, array, col_name, explicit_nulls);
-        }
-        DataType::Null => {
-            if explicit_nulls {
-                rows.iter_mut()
-                    .filter_map(|maybe_row| maybe_row.as_mut())
-                    .for_each(|row| {
-                        row.insert(col_name.to_string(), Value::Null);
-                    });
-            }
-        }
-        DataType::Boolean => {
-            set_column_by_array_type!(as_boolean_array, col_name, rows, array, explicit_nulls);
-        }
-        DataType::Utf8 => {
-            set_column_by_array_type!(as_string_array, col_name, rows, array, explicit_nulls);
-        }
-        DataType::LargeUtf8 => {
-            set_column_by_array_type!(as_largestring_array, col_name, rows, array, explicit_nulls);
-        }
-        DataType::Date32
-        | DataType::Date64
-        | DataType::Timestamp(_, _)
-        | DataType::Time32(_)
-        | DataType::Time64(_)
-        | DataType::Duration(_) => {
-            let options = FormatOptions::default();
-            let formatter = ArrayFormatter::try_new(array.as_ref(), &options)?;
-            let nulls = array.nulls();
-            rows.iter_mut()
-                .enumerate()
-                .filter_map(|(idx, maybe_row)| maybe_row.as_mut().map(|row| (idx, row)))
-                .for_each(|(idx, row)| {
-                    let maybe_value = nulls
-                        .map(|x| x.is_valid(idx))
-                        .unwrap_or(true)
-                        .then(|| formatter.value(idx).to_string().into());
-                    if let Some(j) = maybe_value {
-                        row.insert(col_name.to_string(), j);
-                    } else if explicit_nulls {
-                        row.insert(col_name.to_string(), Value::Null);
-                    }
-                });
-        }
-        DataType::Struct(_) => {
-            let inner_objs = struct_array_to_jsonmap_array(array.as_struct(), explicit_nulls)?;
-            rows.iter_mut()
-                .zip(inner_objs)
-                .filter_map(|(maybe_row, maybe_obj)| maybe_row.as_mut().map(|row| (row, maybe_obj)))
-                .for_each(|(row, maybe_obj)| {
-                    let json = if let Some(obj) = maybe_obj {
-                        Value::Object(obj)
-                    } else {
-                        Value::Null
-                    };
-                    row.insert(col_name.to_string(), json);
-                });
-        }
-        DataType::List(_) => {
-            let listarr = as_list_array(array);
-            rows.iter_mut()
-                .zip(listarr.iter())
-                .filter_map(|(maybe_row, maybe_value)| {
-                    maybe_row.as_mut().map(|row| (row, maybe_value))
-                })
-                .try_for_each(|(row, maybe_value)| -> Result<(), ArrowError> {
-                    let maybe_value = maybe_value
-                        .map(|v| array_to_json_array_internal(&v, explicit_nulls).map(Value::Array))
-                        .transpose()?;
-                    if let Some(j) = maybe_value {
-                        row.insert(col_name.to_string(), j);
-                    } else if explicit_nulls {
-                        row.insert(col_name.to_string(), Value::Null);
-                    }
-                    Ok(())
-                })?;
-        }
-        DataType::LargeList(_) => {
-            let listarr = as_large_list_array(array);
-            rows.iter_mut()
-                .zip(listarr.iter())
-                .filter_map(|(maybe_row, maybe_value)| {
-                    maybe_row.as_mut().map(|row| (row, maybe_value))
-                })
-                .try_for_each(|(row, maybe_value)| -> Result<(), ArrowError> {
-                    let maybe_value = maybe_value
-                        .map(|v| array_to_json_array_internal(&v, explicit_nulls).map(Value::Array))
-                        .transpose()?;
-                    if let Some(j) = maybe_value {
-                        row.insert(col_name.to_string(), j);
-                    } else if explicit_nulls {
-                        row.insert(col_name.to_string(), Value::Null);
-                    }
-                    Ok(())
-                })?;
-        }
-        DataType::Dictionary(_, value_type) => {
-            let hydrated = arrow_cast::cast::cast(&array, value_type)
-                .expect("cannot cast dictionary to underlying values");
-            set_column_for_json_rows(rows, &hydrated, col_name, explicit_nulls)?;
-        }
-        DataType::Map(_, _) => {
-            let maparr = as_map_array(array);
-
-            let keys = maparr.keys();
-            let values = maparr.values();
-
-            // Keys have to be strings to convert to json.
-            if !matches!(keys.data_type(), DataType::Utf8) {
-                return Err(ArrowError::JsonError(format!(
-                    "data type {:?} not supported in nested map for json writer",
-                    keys.data_type()
-                )));
-            }
-
-            let keys = keys.as_string::<i32>();
-            let values = array_to_json_array_internal(values, explicit_nulls)?;
-
-            let mut kv = keys.iter().zip(values);
-
-            for (i, row) in rows
-                .iter_mut()
-                .enumerate()
-                .filter_map(|(i, maybe_row)| maybe_row.as_mut().map(|row| (i, row)))
-            {
-                if maparr.is_null(i) {
-                    row.insert(col_name.to_string(), serde_json::Value::Null);
-                    continue;
-                }
-
-                let len = maparr.value_length(i) as usize;
-                let mut obj = serde_json::Map::new();
-
-                for (_, (k, v)) in (0..len).zip(&mut kv) {
-                    obj.insert(k.expect("keys in a map should be non-null").to_string(), v);
-                }
-
-                row.insert(col_name.to_string(), serde_json::Value::Object(obj));
-            }
-        }
-        _ => {
-            return Err(ArrowError::JsonError(format!(
-                "data type {:?} not supported in nested map for json writer",
-                array.data_type()
-            )))
-        }
-    }
-    Ok(())
-}
-
-/// Converts an arrow [`RecordBatch`] into a `Vec` of Serde JSON
-/// [`JsonMap`]s (objects)
-#[deprecated(note = "Use Writer")]
-pub fn record_batches_to_json_rows(
-    batches: &[&RecordBatch],
-) -> Result<Vec<JsonMap<String, Value>>, ArrowError> {
-    // For backwards compatibility, default to skip nulls
-    record_batches_to_json_rows_internal(batches, false)
-}
-
-fn record_batches_to_json_rows_internal(
-    batches: &[&RecordBatch],
-    explicit_nulls: bool,
-) -> Result<Vec<JsonMap<String, Value>>, ArrowError> {
-    let mut rows: Vec<Option<JsonMap<String, Value>>> = iter::repeat(Some(JsonMap::new()))
-        .take(batches.iter().map(|b| b.num_rows()).sum())
-        .collect();
-
-    if !rows.is_empty() {
-        let schema = batches[0].schema();
-        let mut base = 0;
-        for batch in batches {
-            let row_count = batch.num_rows();
-            let row_slice = &mut rows[base..base + batch.num_rows()];
-            for (j, col) in batch.columns().iter().enumerate() {
-                let col_name = schema.field(j).name();
-                set_column_for_json_rows(row_slice, col, col_name, explicit_nulls)?
-            }
-            base += row_count;
-        }
-    }
-
-    let rows = rows.into_iter().map(|a| a.unwrap()).collect::<Vec<_>>();
-    Ok(rows)
-}
-
 /// This trait defines how to format a sequence of JSON objects to a
 /// byte stream.
 pub trait JsonFormat: Debug + Default {
@@ -731,23 +318,6 @@ where
         }
     }
 
-    /// Write a single JSON row to the output writer
-    #[deprecated(note = "Use Writer::write")]
-    pub fn write_row(&mut self, row: &Value) -> Result<(), ArrowError> {
-        let is_first_row = !self.started;
-        if !self.started {
-            self.format.start_stream(&mut self.writer)?;
-            self.started = true;
-        }
-
-        self.format.start_row(&mut self.writer, is_first_row)?;
-        self.writer.write_all(
-            &serde_json::to_vec(row).map_err(|error| ArrowError::JsonError(error.to_string()))?,
-        )?;
-        self.format.end_row(&mut self.writer)?;
-        Ok(())
-    }
-
     /// Serialize `batch` to JSON output
     pub fn write(&mut self, batch: &RecordBatch) -> Result<(), ArrowError> {
         if batch.num_rows() == 0 {
@@ -831,12 +401,10 @@ mod tests {
     use std::io::{BufReader, Seek};
     use std::sync::Arc;
 
-    use serde_json::json;
+    use serde_json::{json, Value};
 
-    use arrow_array::builder::{
-        FixedSizeBinaryBuilder, FixedSizeListBuilder, Int32Builder, Int64Builder, MapBuilder,
-        StringBuilder,
-    };
+    use arrow_array::builder::*;
+    use arrow_array::types::*;
     use arrow_buffer::{Buffer, NullBuffer, OffsetBuffer, ToByteSlice};
     use arrow_data::ArrayData;
 
@@ -1585,78 +1153,6 @@ mod tests {
         assert_eq!(String::from_utf8(writer.into_inner()).unwrap(), "");
     }
 
-    #[test]
-    #[allow(deprecated)]
-    fn json_writer_one_row() {
-        let mut writer = ArrayWriter::new(vec![] as Vec<u8>);
-        let v = json!({ "an": "object" });
-        writer.write_row(&v).unwrap();
-        writer.finish().unwrap();
-        assert_eq!(
-            String::from_utf8(writer.into_inner()).unwrap(),
-            r#"[{"an":"object"}]"#
-        );
-    }
-
-    #[test]
-    #[allow(deprecated)]
-    fn json_writer_two_rows() {
-        let mut writer = ArrayWriter::new(vec![] as Vec<u8>);
-        let v = json!({ "an": "object" });
-        writer.write_row(&v).unwrap();
-        let v = json!({ "another": "object" });
-        writer.write_row(&v).unwrap();
-        writer.finish().unwrap();
-        assert_eq!(
-            String::from_utf8(writer.into_inner()).unwrap(),
-            r#"[{"an":"object"},{"another":"object"}]"#
-        );
-    }
-
-    #[test]
-    #[allow(deprecated)]
-    fn json_list_roundtrip() {
-        let json_content = r#"
-        {"list": [{"ints": 1}]}
-        {"list": [{}]}
-        {"list": []}
-        {"list": null}
-        {"list": [{"ints": null}]}
-        {"list": [null]}
-        "#;
-        let ints_struct = DataType::Struct(vec![Field::new("ints", DataType::Int32, true)].into());
-        let list_type = DataType::List(Arc::new(Field::new("item", ints_struct, true)));
-        let list_field = Field::new("list", list_type, true);
-        let schema = Arc::new(Schema::new(vec![list_field]));
-        let builder = ReaderBuilder::new(schema).with_batch_size(64);
-        let mut reader = builder.build(std::io::Cursor::new(json_content)).unwrap();
-
-        let batch = reader.next().unwrap().unwrap();
-
-        let list_row = batch.column(0).as_list::<i32>();
-        let values = list_row.values();
-        assert_eq!(values.len(), 4);
-        assert_eq!(values.null_count(), 1);
-
-        // write the batch to JSON, and compare output with input
-        let mut buf = Vec::new();
-        {
-            let mut writer = LineDelimitedWriter::new(&mut buf);
-            writer.write_batches(&[&batch]).unwrap();
-        }
-
-        assert_json_eq(
-            &buf,
-            r#"{"list":[{"ints":1}]}
-{"list":[{}]}
-{"list":[]}
-{}
-{"list":[{}]}
-{"list":[null]}
-"#,
-        );
-    }
-
     #[test]
     fn json_struct_array_nulls() {
         let inner = ListArray::from_iter_primitive::<Int32Type, _, _>(vec![
@@ -1840,78 +1336,6 @@ mod tests {
         }
     }
 
-    #[test]
-    #[allow(deprecated)]
-    fn test_array_to_json_array_for_fixed_size_list_array() {
-        let expected_json = vec![
-            json!([0, 1, 2]),
-            json!(null),
-            json!([3, null, 5]),
-            json!([6, 7, 45]),
-        ];
-
-        let data = vec![
-            Some(vec![Some(0), Some(1), Some(2)]),
-            None,
-            Some(vec![Some(3), None, Some(5)]),
-            Some(vec![Some(6), Some(7), Some(45)]),
-        ];
-
-        let list_array = FixedSizeListArray::from_iter_primitive::<Int32Type, _, _>(data, 3);
-        let list_array = Arc::new(list_array) as ArrayRef;
-
-        assert_eq!(array_to_json_array(&list_array).unwrap(), expected_json);
-    }
-
-    #[test]
-    #[allow(deprecated)]
-    fn test_array_to_json_array_for_map_array() {
-        let expected_json = serde_json::from_value::<Vec<Value>>(json!([
-            [
-                {
-                    "keys": "joe",
-                    "values": 1
-                }
-            ],
-            [
-                {
-                    "keys": "blogs",
-                    "values": 2
-                },
-                {
-                    "keys": "foo",
-                    "values": 4
-                }
-            ],
-            [],
-            null
-        ]))
-        .unwrap();
-
-        let string_builder = StringBuilder::new();
-        let int_builder = Int32Builder::with_capacity(4);
-
-        let mut builder = MapBuilder::new(None, string_builder, int_builder);
-
-        builder.keys().append_value("joe");
-        builder.values().append_value(1);
-        builder.append(true).unwrap();
-
-        builder.keys().append_value("blogs");
-        builder.values().append_value(2);
-        builder.keys().append_value("foo");
-        builder.values().append_value(4);
-        builder.append(true).unwrap();
-        builder.append(true).unwrap();
-        builder.append(false).unwrap();
-
-        let array = builder.finish();
-
-        let map_array = Arc::new(array) as ArrayRef;
-
-        assert_eq!(array_to_json_array(&map_array).unwrap(), expected_json);
-    }
-
     #[test]
     fn test_writer_explicit_nulls() -> Result<(), ArrowError> {
         fn nested_list() -> (Arc<ListArray>, Arc<Field>) {
diff --git a/arrow-ord/src/cmp.rs b/arrow-ord/src/cmp.rs
index bfb1f64e2..fd37da937 100644
--- a/arrow-ord/src/cmp.rs
+++ b/arrow-ord/src/cmp.rs
@@ -62,7 +62,10 @@ impl std::fmt::Display for Op {
     }
 }
 
-/// Perform `left == right` operation on two [`Datum`]
+/// Perform `left == right` operation on two [`Datum`].
+///
+/// Comparing null values on either side will yield a null in the corresponding
+/// slot of the resulting [`BooleanArray`].
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -74,7 +77,10 @@ pub fn eq(lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, ArrowError>
     compare_op(Op::Equal, lhs, rhs)
 }
 
-/// Perform `left != right` operation on two [`Datum`]
+/// Perform `left != right` operation on two [`Datum`].
+///
+/// Comparing null values on either side will yield a null in the corresponding
+/// slot of the resulting [`BooleanArray`].
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -86,7 +92,10 @@ pub fn neq(lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, ArrowError>
     compare_op(Op::NotEqual, lhs, rhs)
 }
 
-/// Perform `left < right` operation on two [`Datum`]
+/// Perform `left < right` operation on two [`Datum`].
+///
+/// Comparing null values on either side will yield a null in the corresponding
+/// slot of the resulting [`BooleanArray`].
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -98,7 +107,10 @@ pub fn lt(lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, ArrowError>
     compare_op(Op::Less, lhs, rhs)
 }
 
-/// Perform `left <= right` operation on two [`Datum`]
+/// Perform `left <= right` operation on two [`Datum`].
+///
+/// Comparing null values on either side will yield a null in the corresponding
+/// slot of the resulting [`BooleanArray`].
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -110,7 +122,10 @@ pub fn lt_eq(lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, ArrowErro
     compare_op(Op::LessEqual, lhs, rhs)
 }
 
-/// Perform `left > right` operation on two [`Datum`]
+/// Perform `left > right` operation on two [`Datum`].
+///
+/// Comparing null values on either side will yield a null in the corresponding
+/// slot of the resulting [`BooleanArray`].
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -122,7 +137,10 @@ pub fn gt(lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, ArrowError>
     compare_op(Op::Greater, lhs, rhs)
 }
 
-/// Perform `left >= right` operation on two [`Datum`]
+/// Perform `left >= right` operation on two [`Datum`].
+///
+/// Comparing null values on either side will yield a null in the corresponding
+/// slot of the resulting [`BooleanArray`].
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
diff --git a/arrow-ord/src/comparison.rs b/arrow-ord/src/comparison.rs
index 4d552b038..1679fae69 100644
--- a/arrow-ord/src/comparison.rs
+++ b/arrow-ord/src/comparison.rs
@@ -1031,8 +1031,7 @@ where
     crate::cmp::neq(&left, &Scalar::new(&right))
 }
 
-/// Perform `left < right` operation on two [`PrimitiveArray`]s. Null values are less than non-null
-/// values.
+/// Perform `left < right` operation on two [`PrimitiveArray`]s.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1052,7 +1051,6 @@ where
 }
 
 /// Perform `left < right` operation on a [`PrimitiveArray`] and a scalar value.
-/// Null values are less than non-null values.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1069,8 +1067,7 @@ where
     crate::cmp::lt(&left, &Scalar::new(&right))
 }
 
-/// Perform `left <= right` operation on two [`PrimitiveArray`]s. Null values are less than non-null
-/// values.
+/// Perform `left <= right` operation on two [`PrimitiveArray`]s.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1090,7 +1087,6 @@ where
 }
 
 /// Perform `left <= right` operation on a [`PrimitiveArray`] and a scalar value.
-/// Null values are less than non-null values.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1110,8 +1106,7 @@ where
     crate::cmp::lt_eq(&left, &Scalar::new(&right))
 }
 
-/// Perform `left > right` operation on two [`PrimitiveArray`]s. Non-null values are greater than null
-/// values.
+/// Perform `left > right` operation on two [`PrimitiveArray`]s.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1131,7 +1126,6 @@ where
 }
 
 /// Perform `left > right` operation on a [`PrimitiveArray`] and a scalar value.
-/// Non-null values are greater than null values.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1148,8 +1142,7 @@ where
     crate::cmp::gt(&left, &Scalar::new(&right))
 }
 
-/// Perform `left >= right` operation on two [`PrimitiveArray`]s. Non-null values are greater than null
-/// values.
+/// Perform `left >= right` operation on two [`PrimitiveArray`]s.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
@@ -1169,7 +1162,6 @@ where
 }
 
 /// Perform `left >= right` operation on a [`PrimitiveArray`] and a scalar value.
-/// Non-null values are greater than null values.
 ///
 /// For floating values like f32 and f64, this comparison produces an ordering in accordance to
 /// the totalOrder predicate as defined in the IEEE 754 (2008 revision) floating point standard.
diff --git a/arrow/Cargo.toml b/arrow/Cargo.toml
index a938d75b1..9d3c431b3 100644
--- a/arrow/Cargo.toml
+++ b/arrow/Cargo.toml
@@ -77,7 +77,7 @@ pyarrow = ["pyo3", "ffi"]
 # but is run as part of our CI checks
 force_validate = ["arrow-data/force_validate"]
 # Enable ffi support
-ffi = ["arrow-schema/ffi", "arrow-data/ffi"]
+ffi = ["arrow-schema/ffi", "arrow-data/ffi", "arrow-array/ffi"]
 chrono-tz = ["arrow-array/chrono-tz"]
 
 [dev-dependencies]
diff --git a/arrow/src/array/ffi.rs b/arrow/src/array/ffi.rs
deleted file mode 100644
index 43f54a038..000000000
--- a/arrow/src/array/ffi.rs
+++ /dev/null
@@ -1,254 +0,0 @@
-// 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.
-
-//! Contains functionality to load an ArrayData from the C Data Interface
-
-use crate::{error::Result, ffi};
-
-use super::ArrayRef;
-
-/// Exports an array to raw pointers of the C Data Interface provided by the consumer.
-/// # Safety
-/// Assumes that these pointers represent valid C Data Interfaces, both in memory
-/// representation and lifetime via the `release` mechanism.
-///
-/// This function copies the content of two FFI structs [ffi::FFI_ArrowArray] and
-/// [ffi::FFI_ArrowSchema] in the array to the location pointed by the raw pointers.
-/// Usually the raw pointers are provided by the array data consumer.
-#[deprecated(note = "Use FFI_ArrowArray::new and FFI_ArrowSchema::try_from")]
-pub unsafe fn export_array_into_raw(
-    src: ArrayRef,
-    out_array: *mut ffi::FFI_ArrowArray,
-    out_schema: *mut ffi::FFI_ArrowSchema,
-) -> Result<()> {
-    let data = src.to_data();
-    let array = ffi::FFI_ArrowArray::new(&data);
-    let schema = ffi::FFI_ArrowSchema::try_from(data.data_type())?;
-
-    std::ptr::write_unaligned(out_array, array);
-    std::ptr::write_unaligned(out_schema, schema);
-
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use crate::array::{DictionaryArray, FixedSizeListArray, Int32Array, StringArray};
-    use crate::buffer::Buffer;
-    use crate::error::Result;
-    use crate::util::bit_util;
-    use crate::{
-        array::{
-            Array, ArrayData, BooleanArray, FixedSizeBinaryArray, Int64Array, StructArray,
-            UInt32Array, UInt64Array,
-        },
-        datatypes::{DataType, Field},
-        ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema},
-    };
-    use std::sync::Arc;
-
-    fn test_round_trip(expected: &ArrayData) -> Result<()> {
-        // here we export the array
-        let array = FFI_ArrowArray::new(expected);
-        let schema = FFI_ArrowSchema::try_from(expected.data_type())?;
-
-        // simulate an external consumer by being the consumer
-        let result = &unsafe { from_ffi(array, &schema) }?;
-
-        assert_eq!(result, expected);
-        Ok(())
-    }
-
-    #[test]
-    fn test_u32() -> Result<()> {
-        let array = UInt32Array::from(vec![Some(2), None, Some(1), None]);
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_u64() -> Result<()> {
-        let array = UInt64Array::from(vec![Some(2), None, Some(1), None]);
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_i64() -> Result<()> {
-        let array = Int64Array::from(vec![Some(2), None, Some(1), None]);
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_struct() -> Result<()> {
-        let inner = StructArray::from(vec![
-            (
-                Arc::new(Field::new("a1", DataType::Boolean, false)),
-                Arc::new(BooleanArray::from(vec![true, true, false, false])) as Arc<dyn Array>,
-            ),
-            (
-                Arc::new(Field::new("a2", DataType::UInt32, false)),
-                Arc::new(UInt32Array::from(vec![1, 2, 3, 4])),
-            ),
-        ]);
-
-        let array = StructArray::from(vec![
-            (
-                Arc::new(Field::new("a", inner.data_type().clone(), false)),
-                Arc::new(inner) as Arc<dyn Array>,
-            ),
-            (
-                Arc::new(Field::new("b", DataType::Boolean, false)),
-                Arc::new(BooleanArray::from(vec![false, false, true, true])) as Arc<dyn Array>,
-            ),
-            (
-                Arc::new(Field::new("c", DataType::UInt32, false)),
-                Arc::new(UInt32Array::from(vec![42, 28, 19, 31])),
-            ),
-        ]);
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_dictionary() -> Result<()> {
-        let values = StringArray::from(vec![Some("foo"), Some("bar"), None]);
-        let keys = Int32Array::from(vec![
-            Some(0),
-            Some(1),
-            None,
-            Some(1),
-            Some(1),
-            None,
-            Some(1),
-            Some(2),
-            Some(1),
-            None,
-        ]);
-        let array = DictionaryArray::new(keys, Arc::new(values));
-
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_fixed_size_binary() -> Result<()> {
-        let values = vec![vec![10, 10, 10], vec![20, 20, 20], vec![30, 30, 30]];
-        let array = FixedSizeBinaryArray::try_from_iter(values.into_iter())?;
-
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_fixed_size_binary_with_nulls() -> Result<()> {
-        let values = vec![
-            None,
-            Some(vec![10, 10, 10]),
-            None,
-            Some(vec![20, 20, 20]),
-            Some(vec![30, 30, 30]),
-            None,
-        ];
-        let array = FixedSizeBinaryArray::try_from_sparse_iter_with_size(values.into_iter(), 3)?;
-
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_fixed_size_list() -> Result<()> {
-        let v: Vec<i64> = (0..9).collect();
-        let value_data = ArrayData::builder(DataType::Int64)
-            .len(9)
-            .add_buffer(Buffer::from_slice_ref(v))
-            .build()?;
-        let list_data_type =
-            DataType::FixedSizeList(Arc::new(Field::new("f", DataType::Int64, false)), 3);
-        let list_data = ArrayData::builder(list_data_type)
-            .len(3)
-            .add_child_data(value_data)
-            .build()?;
-        let array = FixedSizeListArray::from(list_data);
-
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_fixed_size_list_with_nulls() -> Result<()> {
-        // 0100 0110
-        let mut validity_bits: [u8; 1] = [0; 1];
-        bit_util::set_bit(&mut validity_bits, 1);
-        bit_util::set_bit(&mut validity_bits, 2);
-        bit_util::set_bit(&mut validity_bits, 6);
-
-        let v: Vec<i16> = (0..16).collect();
-        let value_data = ArrayData::builder(DataType::Int16)
-            .len(16)
-            .add_buffer(Buffer::from_slice_ref(v))
-            .build()?;
-        let list_data_type =
-            DataType::FixedSizeList(Arc::new(Field::new("f", DataType::Int16, false)), 2);
-        let list_data = ArrayData::builder(list_data_type)
-            .len(8)
-            .null_bit_buffer(Some(Buffer::from(validity_bits)))
-            .add_child_data(value_data)
-            .build()?;
-        let array = FixedSizeListArray::from(list_data);
-
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-
-    #[test]
-    fn test_fixed_size_list_nested() -> Result<()> {
-        let v: Vec<i32> = (0..16).collect();
-        let value_data = ArrayData::builder(DataType::Int32)
-            .len(16)
-            .add_buffer(Buffer::from_slice_ref(v))
-            .build()?;
-
-        let offsets: Vec<i32> = vec![0, 2, 4, 6, 8, 10, 12, 14, 16];
-        let value_offsets = Buffer::from_slice_ref(offsets);
-        let inner_list_data_type =
-            DataType::List(Arc::new(Field::new("item", DataType::Int32, false)));
-        let inner_list_data = ArrayData::builder(inner_list_data_type.clone())
-            .len(8)
-            .add_buffer(value_offsets)
-            .add_child_data(value_data)
-            .build()?;
-
-        // 0000 0100
-        let mut validity_bits: [u8; 1] = [0; 1];
-        bit_util::set_bit(&mut validity_bits, 2);
-
-        let list_data_type =
-            DataType::FixedSizeList(Arc::new(Field::new("f", inner_list_data_type, false)), 2);
-        let list_data = ArrayData::builder(list_data_type)
-            .len(4)
-            .null_bit_buffer(Some(Buffer::from(validity_bits)))
-            .add_child_data(inner_list_data)
-            .build()?;
-
-        let array = FixedSizeListArray::from(list_data);
-
-        let data = array.into_data();
-        test_round_trip(&data)
-    }
-}
diff --git a/arrow/src/array/mod.rs b/arrow/src/array/mod.rs
index fa01f4c4c..b563c320b 100644
--- a/arrow/src/array/mod.rs
+++ b/arrow/src/array/mod.rs
@@ -19,9 +19,6 @@
 //!
 //! **See [arrow_array] for examples and usage instructions**
 
-#[cfg(feature = "ffi")]
-mod ffi;
-
 // --------------------- Array & ArrayData ---------------------
 pub use arrow_array::builder::*;
 pub use arrow_array::cast::*;
@@ -35,7 +32,7 @@ pub use arrow_data::transform::{Capacities, MutableArrayData};
 
 #[cfg(feature = "ffi")]
 #[allow(deprecated)]
-pub use self::ffi::export_array_into_raw;
+pub use arrow_array::ffi::export_array_into_raw;
 
 // --------------------- Array's values comparison ---------------------
 
diff --git a/arrow/src/lib.rs b/arrow/src/lib.rs
index 78e2363e4..09d6fc48a 100644
--- a/arrow/src/lib.rs
+++ b/arrow/src/lib.rs
@@ -363,9 +363,9 @@ pub use arrow_csv as csv;
 pub mod datatypes;
 pub mod error;
 #[cfg(feature = "ffi")]
-pub mod ffi;
+pub use arrow_array::ffi;
 #[cfg(feature = "ffi")]
-pub mod ffi_stream;
+pub use arrow_array::ffi_stream;
 #[cfg(feature = "ipc")]
 pub use arrow_ipc as ipc;
 #[cfg(feature = "json")]
diff --git a/object_store/src/attributes.rs b/object_store/src/attributes.rs
index 9b90b5325..ecef32eb4 100644
--- a/object_store/src/attributes.rs
+++ b/object_store/src/attributes.rs
@@ -23,6 +23,18 @@ use std::ops::Deref;
 #[non_exhaustive]
 #[derive(Debug, Hash, Eq, PartialEq, Clone)]
 pub enum Attribute {
+    /// Specifies how the object should be handled by a browser
+    ///
+    /// See [Content-Disposition](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Disposition)
+    ContentDisposition,
+    /// Specifies the encodings applied to the object
+    ///
+    /// See [Content-Encoding](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Encoding)
+    ContentEncoding,
+    /// Specifies the language of the object
+    ///
+    /// See [Content-Language](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Language)
+    ContentLanguage,
     /// Specifies the MIME type of the object
     ///
     /// This takes precedence over any [ClientOptions](crate::ClientOptions) configuration
@@ -177,12 +189,15 @@ mod tests {
     #[test]
     fn test_attributes_basic() {
         let mut attributes = Attributes::from_iter([
+            (Attribute::ContentDisposition, "inline"),
+            (Attribute::ContentEncoding, "gzip"),
+            (Attribute::ContentLanguage, "en-US"),
             (Attribute::ContentType, "test"),
             (Attribute::CacheControl, "control"),
         ]);
 
         assert!(!attributes.is_empty());
-        assert_eq!(attributes.len(), 2);
+        assert_eq!(attributes.len(), 5);
 
         assert_eq!(
             attributes.get(&Attribute::ContentType),
@@ -195,17 +210,30 @@ mod tests {
             attributes.insert(Attribute::CacheControl, "v1".into()),
             Some(metav)
         );
-        assert_eq!(attributes.len(), 2);
+        assert_eq!(attributes.len(), 5);
 
         assert_eq!(
             attributes.remove(&Attribute::CacheControl).unwrap(),
             "v1".into()
         );
-        assert_eq!(attributes.len(), 1);
+        assert_eq!(attributes.len(), 4);
 
         let metav: AttributeValue = "v2".into();
         attributes.insert(Attribute::CacheControl, metav.clone());
         assert_eq!(attributes.get(&Attribute::CacheControl), Some(&metav));
-        assert_eq!(attributes.len(), 2);
+        assert_eq!(attributes.len(), 5);
+
+        assert_eq!(
+            attributes.get(&Attribute::ContentDisposition),
+            Some(&"inline".into())
+        );
+        assert_eq!(
+            attributes.get(&Attribute::ContentEncoding),
+            Some(&"gzip".into())
+        );
+        assert_eq!(
+            attributes.get(&Attribute::ContentLanguage),
+            Some(&"en-US".into())
+        );
     }
 }
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 4a4dc178d..24247688e 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -42,14 +42,17 @@ use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::{Buf, Bytes};
-use hyper::header::{CACHE_CONTROL, CONTENT_LENGTH};
+use hyper::header::{
+    CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH,
+    CONTENT_TYPE,
+};
 use hyper::http::HeaderName;
 use hyper::{http, HeaderMap};
 use itertools::Itertools;
 use md5::{Digest, Md5};
 use percent_encoding::{utf8_percent_encode, PercentEncode};
 use quick_xml::events::{self as xml_events};
-use reqwest::{header::CONTENT_TYPE, Client as ReqwestClient, Method, RequestBuilder, Response};
+use reqwest::{Client as ReqwestClient, Method, RequestBuilder, Response};
 use ring::digest;
 use ring::digest::Context;
 use serde::{Deserialize, Serialize};
@@ -322,6 +325,9 @@ impl<'a> Request<'a> {
         for (k, v) in &attributes {
             builder = match k {
                 Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()),
+                Attribute::ContentDisposition => builder.header(CONTENT_DISPOSITION, v.as_ref()),
+                Attribute::ContentEncoding => builder.header(CONTENT_ENCODING, v.as_ref()),
+                Attribute::ContentLanguage => builder.header(CONTENT_LANGUAGE, v.as_ref()),
                 Attribute::ContentType => {
                     has_content_type = true;
                     builder.header(CONTENT_TYPE, v.as_ref())
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 7f1edf12f..5bc6d56e7 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -437,10 +437,16 @@ mod tests {
 
         // Object tagging is not supported by S3 Express One Zone
         if config.session_provider.is_none() {
-            tagging(&integration, !config.disable_tagging, |p| {
-                let client = Arc::clone(&integration.client);
-                async move { client.get_object_tagging(&p).await }
-            })
+            tagging(
+                Arc::new(AmazonS3 {
+                    client: Arc::clone(&integration.client),
+                }),
+                !config.disable_tagging,
+                |p| {
+                    let client = Arc::clone(&integration.client);
+                    async move { client.get_object_tagging(&p).await }
+                },
+            )
             .await;
         }
 
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index 918fcd047..311bd72ff 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -50,6 +50,10 @@ use url::Url;
 const VERSION_HEADER: &str = "x-ms-version-id";
 static MS_CACHE_CONTROL: HeaderName = HeaderName::from_static("x-ms-blob-cache-control");
 static MS_CONTENT_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-content-type");
+static MS_CONTENT_DISPOSITION: HeaderName =
+    HeaderName::from_static("x-ms-blob-content-disposition");
+static MS_CONTENT_ENCODING: HeaderName = HeaderName::from_static("x-ms-blob-content-encoding");
+static MS_CONTENT_LANGUAGE: HeaderName = HeaderName::from_static("x-ms-blob-content-language");
 
 static TAGS_HEADER: HeaderName = HeaderName::from_static("x-ms-tags");
 
@@ -206,6 +210,11 @@ impl<'a> PutRequest<'a> {
         for (k, v) in &attributes {
             builder = match k {
                 Attribute::CacheControl => builder.header(&MS_CACHE_CONTROL, v.as_ref()),
+                Attribute::ContentDisposition => {
+                    builder.header(&MS_CONTENT_DISPOSITION, v.as_ref())
+                }
+                Attribute::ContentEncoding => builder.header(&MS_CONTENT_ENCODING, v.as_ref()),
+                Attribute::ContentLanguage => builder.header(&MS_CONTENT_LANGUAGE, v.as_ref()),
                 Attribute::ContentType => {
                     has_content_type = true;
                     builder.header(&MS_CONTENT_TYPE, v.as_ref())
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index 25ae6dda6..755f3b126 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -296,10 +296,16 @@ mod tests {
         signing(&integration).await;
 
         let validate = !integration.client.config().disable_tagging;
-        tagging(&integration, validate, |p| {
-            let client = Arc::clone(&integration.client);
-            async move { client.get_blob_tagging(&p).await }
-        })
+        tagging(
+            Arc::new(MicrosoftAzure {
+                client: Arc::clone(&integration.client),
+            }),
+            validate,
+            |p| {
+                let client = Arc::clone(&integration.client);
+                async move { client.get_blob_tagging(&p).await }
+            },
+        )
         .await;
 
         // Azurite doesn't support attributes properly
diff --git a/object_store/src/buffered.rs b/object_store/src/buffered.rs
index d41224177..feb84d4d0 100644
--- a/object_store/src/buffered.rs
+++ b/object_store/src/buffered.rs
@@ -18,7 +18,10 @@
 //! Utilities for performing tokio-style buffered IO
 
 use crate::path::Path;
-use crate::{ObjectMeta, ObjectStore, PutPayloadMut, WriteMultipart};
+use crate::{
+    Attributes, ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayloadMut, TagSet,
+    WriteMultipart,
+};
 use bytes::Bytes;
 use futures::future::{BoxFuture, FutureExt};
 use futures::ready;
@@ -217,6 +220,8 @@ impl AsyncBufRead for BufReader {
 pub struct BufWriter {
     capacity: usize,
     max_concurrency: usize,
+    attributes: Option<Attributes>,
+    tags: Option<TagSet>,
     state: BufWriterState,
     store: Arc<dyn ObjectStore>,
 }
@@ -252,6 +257,8 @@ impl BufWriter {
             capacity,
             store,
             max_concurrency: 8,
+            attributes: None,
+            tags: None,
             state: BufWriterState::Buffer(path, PutPayloadMut::new()),
         }
     }
@@ -266,6 +273,22 @@ impl BufWriter {
         }
     }
 
+    /// Set the attributes of the uploaded object
+    pub fn with_attributes(self, attributes: Attributes) -> Self {
+        Self {
+            attributes: Some(attributes),
+            ..self
+        }
+    }
+
+    /// Set the tags of the uploaded object
+    pub fn with_tags(self, tags: TagSet) -> Self {
+        Self {
+            tags: Some(tags),
+            ..self
+        }
+    }
+
     /// Abort this writer, cleaning up any partially uploaded state
     ///
     /// # Panic
@@ -306,9 +329,13 @@ impl AsyncWrite for BufWriter {
                     if b.content_length().saturating_add(buf.len()) >= cap {
                         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 store = Arc::clone(&self.store);
                         self.state = BufWriterState::Prepare(Box::pin(async move {
-                            let upload = store.put_multipart(&path).await?;
+                            let upload = store.put_multipart_opts(&path, opts).await?;
                             let mut chunked = WriteMultipart::new_with_chunk_size(upload, cap);
                             for chunk in buffer.freeze() {
                                 chunked.put(chunk);
@@ -346,9 +373,14 @@ impl AsyncWrite for BufWriter {
                 BufWriterState::Buffer(p, b) => {
                     let buf = std::mem::take(b);
                     let path = std::mem::take(p);
+                    let opts = PutOptions {
+                        attributes: self.attributes.take().unwrap_or_default(),
+                        tags: self.tags.take().unwrap_or_default(),
+                        ..Default::default()
+                    };
                     let store = Arc::clone(&self.store);
                     self.state = BufWriterState::Flush(Box::pin(async move {
-                        store.put(&path, buf.into()).await?;
+                        store.put_opts(&path, buf.into(), opts).await?;
                         Ok(())
                     }));
                 }
@@ -383,6 +415,7 @@ mod tests {
     use super::*;
     use crate::memory::InMemory;
     use crate::path::Path;
+    use crate::{Attribute, GetOptions};
     use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
 
     #[tokio::test]
@@ -464,26 +497,54 @@ mod tests {
         }
     }
 
+    // Note: `BufWriter::with_tags` functionality is tested in `crate::tests::tagging`
     #[tokio::test]
     async fn test_buf_writer() {
         let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
         let path = Path::from("file.txt");
+        let attributes = Attributes::from_iter([
+            (Attribute::ContentType, "text/html"),
+            (Attribute::CacheControl, "max-age=604800"),
+        ]);
 
         // Test put
-        let mut writer = BufWriter::with_capacity(Arc::clone(&store), path.clone(), 30);
+        let mut writer = BufWriter::with_capacity(Arc::clone(&store), path.clone(), 30)
+            .with_attributes(attributes.clone());
         writer.write_all(&[0; 20]).await.unwrap();
         writer.flush().await.unwrap();
         writer.write_all(&[0; 5]).await.unwrap();
         writer.shutdown().await.unwrap();
-        assert_eq!(store.head(&path).await.unwrap().size, 25);
+        let response = store
+            .get_opts(
+                &path,
+                GetOptions {
+                    head: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.meta.size, 25);
+        assert_eq!(response.attributes, attributes);
 
         // Test multipart
-        let mut writer = BufWriter::with_capacity(Arc::clone(&store), path.clone(), 30);
+        let mut writer = BufWriter::with_capacity(Arc::clone(&store), path.clone(), 30)
+            .with_attributes(attributes.clone());
         writer.write_all(&[0; 20]).await.unwrap();
         writer.flush().await.unwrap();
         writer.write_all(&[0; 20]).await.unwrap();
         writer.shutdown().await.unwrap();
-
-        assert_eq!(store.head(&path).await.unwrap().size, 40);
+        let response = store
+            .get_opts(
+                &path,
+                GetOptions {
+                    head: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.meta.size, 40);
+        assert_eq!(response.attributes, attributes);
     }
 }
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index f70045761..430b87b1b 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -22,7 +22,10 @@ use crate::path::Path;
 use crate::{Attribute, Attributes, GetOptions, GetRange, GetResult, GetResultPayload, Result};
 use async_trait::async_trait;
 use futures::{StreamExt, TryStreamExt};
-use hyper::header::{CACHE_CONTROL, CONTENT_RANGE, CONTENT_TYPE};
+use hyper::header::{
+    CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_RANGE,
+    CONTENT_TYPE,
+};
 use hyper::StatusCode;
 use reqwest::header::ToStrError;
 use reqwest::Response;
@@ -120,6 +123,15 @@ enum GetResultError {
     #[snafu(display("Cache-Control header contained non UTF-8 characters"))]
     InvalidCacheControl { source: ToStrError },
 
+    #[snafu(display("Content-Disposition header contained non UTF-8 characters"))]
+    InvalidContentDisposition { source: ToStrError },
+
+    #[snafu(display("Content-Encoding header contained non UTF-8 characters"))]
+    InvalidContentEncoding { source: ToStrError },
+
+    #[snafu(display("Content-Language header contained non UTF-8 characters"))]
+    InvalidContentLanguage { source: ToStrError },
+
     #[snafu(display("Content-Type header contained non UTF-8 characters"))]
     InvalidContentType { source: ToStrError },
 
@@ -167,16 +179,48 @@ fn get_result<T: GetClient>(
         0..meta.size
     };
 
-    let mut attributes = Attributes::new();
-    if let Some(x) = response.headers().get(CACHE_CONTROL) {
-        let x = x.to_str().context(InvalidCacheControlSnafu)?;
-        attributes.insert(Attribute::CacheControl, x.to_string().into());
-    }
-    if let Some(x) = response.headers().get(CONTENT_TYPE) {
-        let x = x.to_str().context(InvalidContentTypeSnafu)?;
-        attributes.insert(Attribute::ContentType, x.to_string().into());
+    macro_rules! parse_attributes {
+        ($headers:expr, $(($header:expr, $attr:expr, $err:expr)),*) => {{
+            let mut attributes = Attributes::new();
+            $(
+            if let Some(x) = $headers.get($header) {
+                let x = x.to_str().context($err)?;
+                attributes.insert($attr, x.to_string().into());
+            }
+            )*
+            attributes
+        }}
     }
 
+    let attributes = parse_attributes!(
+        response.headers(),
+        (
+            CACHE_CONTROL,
+            Attribute::CacheControl,
+            InvalidCacheControlSnafu
+        ),
+        (
+            CONTENT_DISPOSITION,
+            Attribute::ContentDisposition,
+            InvalidContentDispositionSnafu
+        ),
+        (
+            CONTENT_ENCODING,
+            Attribute::ContentEncoding,
+            InvalidContentEncodingSnafu
+        ),
+        (
+            CONTENT_LANGUAGE,
+            Attribute::ContentLanguage,
+            InvalidContentLanguageSnafu
+        ),
+        (
+            CONTENT_TYPE,
+            Attribute::ContentType,
+            InvalidContentTypeSnafu
+        )
+    );
+
     let stream = response
         .bytes_stream()
         .map_err(|source| crate::Error::Generic {
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index 9c39efe6b..a54932565 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -36,7 +36,10 @@ use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::Buf;
-use hyper::header::{CACHE_CONTROL, CONTENT_LENGTH, CONTENT_TYPE};
+use hyper::header::{
+    CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH,
+    CONTENT_TYPE,
+};
 use percent_encoding::{percent_encode, utf8_percent_encode, NON_ALPHANUMERIC};
 use reqwest::header::HeaderName;
 use reqwest::{Client, Method, RequestBuilder, Response, StatusCode};
@@ -195,6 +198,9 @@ impl<'a> Request<'a> {
         for (k, v) in &attributes {
             builder = match k {
                 Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()),
+                Attribute::ContentDisposition => builder.header(CONTENT_DISPOSITION, v.as_ref()),
+                Attribute::ContentEncoding => builder.header(CONTENT_ENCODING, v.as_ref()),
+                Attribute::ContentLanguage => builder.header(CONTENT_LANGUAGE, v.as_ref()),
                 Attribute::ContentType => {
                     has_content_type = true;
                     builder.header(CONTENT_TYPE, v.as_ref())
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index cf259196b..4dccef880 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -25,9 +25,11 @@ use crate::{Attribute, Attributes, ClientOptions, GetOptions, ObjectMeta, PutPay
 use async_trait::async_trait;
 use bytes::Buf;
 use chrono::{DateTime, Utc};
-use hyper::header::{CACHE_CONTROL, CONTENT_LENGTH};
+use hyper::header::{
+    CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH,
+    CONTENT_TYPE,
+};
 use percent_encoding::percent_decode_str;
-use reqwest::header::CONTENT_TYPE;
 use reqwest::{Method, Response, StatusCode};
 use serde::Deserialize;
 use snafu::{OptionExt, ResultExt, Snafu};
@@ -172,6 +174,11 @@ impl Client {
             for (k, v) in &attributes {
                 builder = match k {
                     Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()),
+                    Attribute::ContentDisposition => {
+                        builder.header(CONTENT_DISPOSITION, v.as_ref())
+                    }
+                    Attribute::ContentEncoding => builder.header(CONTENT_ENCODING, v.as_ref()),
+                    Attribute::ContentLanguage => builder.header(CONTENT_LANGUAGE, v.as_ref()),
                     Attribute::ContentType => {
                         has_content_type = true;
                         builder.header(CONTENT_TYPE, v.as_ref())
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index ad72bd29e..9a8f77b4d 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -1314,12 +1314,14 @@ mod test_util {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::buffered::BufWriter;
     use crate::multipart::MultipartStore;
     use crate::test_util::flatten_list_stream;
     use chrono::TimeZone;
     use futures::stream::FuturesUnordered;
     use rand::distributions::Alphanumeric;
     use rand::{thread_rng, Rng};
+    use tokio::io::AsyncWriteExt;
 
     pub(crate) async fn put_get_delete_list(storage: &DynObjectStore) {
         delete_fixtures(storage).await;
@@ -1744,8 +1746,14 @@ mod tests {
     pub(crate) async fn put_get_attributes(integration: &dyn ObjectStore) {
         // Test handling of attributes
         let attributes = Attributes::from_iter([
-            (Attribute::ContentType, "text/html; charset=utf-8"),
             (Attribute::CacheControl, "max-age=604800"),
+            (
+                Attribute::ContentDisposition,
+                r#"attachment; filename="test.html""#,
+            ),
+            (Attribute::ContentEncoding, "gzip"),
+            (Attribute::ContentLanguage, "en-US"),
+            (Attribute::ContentType, "text/html; charset=utf-8"),
         ]);
 
         let path = Path::from("attributes");
@@ -2359,7 +2367,7 @@ mod tests {
     }
 
     #[cfg(any(feature = "aws", feature = "azure"))]
-    pub(crate) async fn tagging<F, Fut>(storage: &dyn ObjectStore, validate: bool, get_tags: F)
+    pub(crate) async fn tagging<F, Fut>(storage: Arc<dyn ObjectStore>, validate: bool, get_tags: F)
     where
         F: Fn(Path) -> Fut + Send + Sync,
         Fut: std::future::Future<Output = Result<reqwest::Response>> + Send,
@@ -2409,19 +2417,24 @@ mod tests {
 
         let multi_path = Path::from("tag_test_multi");
         let mut write = storage
-            .put_multipart_opts(&multi_path, tag_set.into())
+            .put_multipart_opts(&multi_path, tag_set.clone().into())
             .await
             .unwrap();
 
         write.put_part("foo".into()).await.unwrap();
         write.complete().await.unwrap();
 
+        let buf_path = Path::from("tag_test_buf");
+        let mut buf = BufWriter::new(storage, buf_path.clone()).with_tags(tag_set);
+        buf.write_all(b"foo").await.unwrap();
+        buf.shutdown().await.unwrap();
+
         // Write should always succeed, but certain configurations may simply ignore tags
         if !validate {
             return;
         }
 
-        for path in [path, multi_path] {
+        for path in [path, multi_path, buf_path] {
             let resp = get_tags(path.clone()).await.unwrap();
             let body = resp.bytes().await.unwrap();
 
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index 4fcf855d1..348ac5c90 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -44,7 +44,7 @@ arrow-schema = { workspace = true, optional = true }
 arrow-select = { workspace = true, optional = true }
 arrow-ipc = { workspace = true, optional = true }
 # Intentionally not a path dependency as object_store is released separately
-object_store = { version = "0.9.0", default-features = false, optional = true }
+object_store = { version = "0.10.0", default-features = false, optional = true }
 
 bytes = { version = "1.1", default-features = false, features = ["std"] }
 thrift = { version = "0.17", default-features = false }
@@ -87,7 +87,7 @@ serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt", "io-util", "fs"] }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
-object_store = { version = "0.9.0", default-features = false, features = ["azure"] }
+object_store = { version = "0.10.0", default-features = false, features = ["azure"] }
 
 [package.metadata.docs.rs]
 all-features = true


Test patch:
diff --git a/arrow/tests/array_validation.rs b/arrow/tests/array_validation.rs
index f5298f82e..41def9051 100644
--- a/arrow/tests/array_validation.rs
+++ b/arrow/tests/array_validation.rs
@@ -342,6 +342,94 @@ fn test_validate_offsets_last_too_large() {
     .unwrap();
 }
 
+/// Test that the list of type `data_type` generates correct offset and size out of bounds errors
+fn check_list_view_offsets_sizes<T: ArrowNativeType>(
+    data_type: DataType,
+    offsets: Vec<T>,
+    sizes: Vec<T>,
+) {
+    let values: Int32Array = [Some(1), Some(2), Some(3), Some(4)].into_iter().collect();
+    let offsets_buffer = Buffer::from_slice_ref(offsets);
+    let sizes_buffer = Buffer::from_slice_ref(sizes);
+    ArrayData::try_new(
+        data_type,
+        4,
+        None,
+        0,
+        vec![offsets_buffer, sizes_buffer],
+        vec![values.into_data()],
+    )
+    .unwrap();
+}
+
+#[test]
+#[should_panic(expected = "Size 3 at index 3 is larger than the remaining values for ListView")]
+fn test_validate_list_view_offsets_sizes() {
+    let field_type = Field::new("f", DataType::Int32, true);
+    check_list_view_offsets_sizes::<i32>(
+        DataType::ListView(Arc::new(field_type)),
+        vec![0, 1, 1, 2],
+        vec![1, 1, 1, 3],
+    );
+}
+
+#[test]
+#[should_panic(
+    expected = "Size 3 at index 3 is larger than the remaining values for LargeListView"
+)]
+fn test_validate_large_list_view_offsets_sizes() {
+    let field_type = Field::new("f", DataType::Int32, true);
+    check_list_view_offsets_sizes::<i64>(
+        DataType::LargeListView(Arc::new(field_type)),
+        vec![0, 1, 1, 2],
+        vec![1, 1, 1, 3],
+    );
+}
+
+#[test]
+#[should_panic(expected = "Error converting offset[1] (-1) to usize for ListView")]
+fn test_validate_list_view_negative_offsets() {
+    let field_type = Field::new("f", DataType::Int32, true);
+    check_list_view_offsets_sizes::<i32>(
+        DataType::ListView(Arc::new(field_type)),
+        vec![0, -1, 1, 2],
+        vec![1, 1, 1, 3],
+    );
+}
+
+#[test]
+#[should_panic(expected = "Error converting size[2] (-1) to usize for ListView")]
+fn test_validate_list_view_negative_sizes() {
+    let field_type = Field::new("f", DataType::Int32, true);
+    check_list_view_offsets_sizes::<i32>(
+        DataType::ListView(Arc::new(field_type)),
+        vec![0, 1, 1, 2],
+        vec![1, 1, -1, 3],
+    );
+}
+
+#[test]
+#[should_panic(expected = "Error converting offset[1] (-1) to usize for LargeListView")]
+fn test_validate_large_list_view_negative_offsets() {
+    let field_type = Field::new("f", DataType::Int32, true);
+    check_list_view_offsets_sizes::<i64>(
+        DataType::LargeListView(Arc::new(field_type)),
+        vec![0, -1, 1, 2],
+        vec![1, 1, 1, 3],
+    );
+}
+
+#[test]
+#[should_panic(expected = "Error converting size[2] (-1) to usize for LargeListView")]
+fn test_validate_large_list_view_negative_sizes() {
+    let field_type = Field::new("f", DataType::Int32, true);
+    check_list_view_offsets_sizes::<i64>(
+        DataType::LargeListView(Arc::new(field_type)),
+        vec![0, 1, 1, 2],
+        vec![1, 1, -1, 3],
+    );
+}
+
 #[test]
 #[should_panic(
     expected = "Values length 4 is less than the length (2) multiplied by the value size (2) for FixedSizeList"