Repository: apache/arrow-rs
Pull request #5651 (2024-04-16 09:35:43)

Remove deprecated JSON writer
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes #.

# 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.
-->

These APIs were deprecated in #5318, have fairly limited test coverage now that the main writer doesn't use them, and have patchy support for nested types. They can now be removed.

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->



Patch:
diff --git a/.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/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-flight/gen/Cargo.toml b/arrow-flight/gen/Cargo.toml
index 56269ee25..9547fd6df 100644
--- a/arrow-flight/gen/Cargo.toml
+++ b/arrow-flight/gen/Cargo.toml
@@ -32,6 +32,6 @@ publish = false
 [dependencies]
 # Pin specific version of the tonic-build dependencies to avoid auto-generated
 # (and checked in) arrow.flight.protocol.rs from changing
-proc-macro2 = { version = "=1.0.80", default-features = false }
+proc-macro2 = { version = "=1.0.81", default-features = false }
 prost-build = { version = "=0.12.4", default-features = false }
 tonic-build = { version = "=0.11.0", default-features = false, features = ["transport", "prost"] }
diff --git a/arrow-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/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/CHANGELOG-old.md b/object_store/CHANGELOG-old.md
index 141a8b9ff..3ccfcad01 100644
--- a/object_store/CHANGELOG-old.md
+++ b/object_store/CHANGELOG-old.md
@@ -21,6 +21,48 @@
 
 # Changelog
 
+## [object_store_0.9.1](https://github.com/apache/arrow-rs/tree/object_store_0.9.1) (2024-03-01)
+
+[Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.9.0...object_store_0.9.1)
+
+**Implemented enhancements:**
+
+- \[object\_store\] Enable anonymous read access for Azure [\#5424](https://github.com/apache/arrow-rs/issues/5424) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Support for additional URL formats in object\_store for Azure blob [\#5370](https://github.com/apache/arrow-rs/issues/5370) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Mention "Http" support in README [\#5320](https://github.com/apache/arrow-rs/issues/5320) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Pass Options to HttpBuilder in parse\_url\_opts [\#5310](https://github.com/apache/arrow-rs/issues/5310) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Remove Localstack DynamoDB Workaround Once Fixed Upstream [\#5267](https://github.com/apache/arrow-rs/issues/5267) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Can I use S3 server side encryption [\#5087](https://github.com/apache/arrow-rs/issues/5087) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+
+**Fixed bugs:**
+
+- delete\_stream fails in MinIO [\#5414](https://github.com/apache/arrow-rs/issues/5414) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- \[object\_store\] Completing an empty Multipart Upload fails for AWS S3 [\#5404](https://github.com/apache/arrow-rs/issues/5404) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Multipart upload can leave futures unpolled, leading to timeout [\#5366](https://github.com/apache/arrow-rs/issues/5366) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Broken Link in README \(Rust Object Store\) Content [\#5309](https://github.com/apache/arrow-rs/issues/5309) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+
+**Merged pull requests:**
+
+- Expose path\_to\_filesystem public [\#5441](https://github.com/apache/arrow-rs/pull/5441) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([metesynnada](https://github.com/metesynnada))
+- Update nix requirement from 0.27.1 to 0.28.0 in /object\_store [\#5432](https://github.com/apache/arrow-rs/pull/5432) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- Add BufWriter for Adapative Put / Multipart Upload [\#5431](https://github.com/apache/arrow-rs/pull/5431) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Enable anonymous access for MicrosoftAzure [\#5425](https://github.com/apache/arrow-rs/pull/5425) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([andrebsguedes](https://github.com/andrebsguedes))
+- fix\(object\_store\): Include Content-MD5 header for S3 DeleteObjects [\#5415](https://github.com/apache/arrow-rs/pull/5415) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([paraseba](https://github.com/paraseba))
+- docds\(object\_store\): Mention HTTP/WebDAV in README [\#5409](https://github.com/apache/arrow-rs/pull/5409) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([Xuanwo](https://github.com/Xuanwo))
+- \[object\_store\] Fix empty Multipart Upload for AWS S3 [\#5405](https://github.com/apache/arrow-rs/pull/5405) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([andrebsguedes](https://github.com/andrebsguedes))
+- feat: S3 server-side encryption [\#5402](https://github.com/apache/arrow-rs/pull/5402) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([wjones127](https://github.com/wjones127))
+- Pull container name from URL for Azure blob [\#5371](https://github.com/apache/arrow-rs/pull/5371) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([bradvoth](https://github.com/bradvoth))
+- docs\(object-store\): add warning to flush [\#5369](https://github.com/apache/arrow-rs/pull/5369) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([wjones127](https://github.com/wjones127))
+- Minor\(docs\): update master to main for DataFusion/Ballista [\#5363](https://github.com/apache/arrow-rs/pull/5363) ([caicancai](https://github.com/caicancai))
+- Test parse\_url\_opts for HTTP \(\#5310\) [\#5316](https://github.com/apache/arrow-rs/pull/5316) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Update IOx links [\#5312](https://github.com/apache/arrow-rs/pull/5312) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Pass options to HTTPBuilder in parse\_url\_opts \(\#5310\) [\#5311](https://github.com/apache/arrow-rs/pull/5311) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Bump actions/cache from 3 to 4 [\#5308](https://github.com/apache/arrow-rs/pull/5308) ([dependabot[bot]](https://github.com/apps/dependabot))
+- Remove localstack DynamoDB workaround \(\#5267\) [\#5307](https://github.com/apache/arrow-rs/pull/5307) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- refactor: log server error during object store retries [\#5294](https://github.com/apache/arrow-rs/pull/5294) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([crepererum](https://github.com/crepererum))
+- Prepare arrow 50.0.0 [\#5291](https://github.com/apache/arrow-rs/pull/5291) ([tustvold](https://github.com/tustvold))
+- Enable JS tests again [\#5287](https://github.com/apache/arrow-rs/pull/5287) ([domoritz](https://github.com/domoritz))
+
 ## [object_store_0.9.0](https://github.com/apache/arrow-rs/tree/object_store_0.9.0) (2024-01-05)
 
 [Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.8.0...object_store_0.9.0)
diff --git a/object_store/CHANGELOG.md b/object_store/CHANGELOG.md
index 0c6af6766..dc58ecb0a 100644
--- a/object_store/CHANGELOG.md
+++ b/object_store/CHANGELOG.md
@@ -19,46 +19,58 @@
 
 # Changelog
 
-## [object_store_0.9.1](https://github.com/apache/arrow-rs/tree/object_store_0.9.1) (2024-03-01)
+## [object_store_0.10.0](https://github.com/apache/arrow-rs/tree/object_store_0.10.0) (2024-04-17)
 
-[Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.9.0...object_store_0.9.1)
+[Full Changelog](https://github.com/apache/arrow-rs/compare/object_store_0.9.1...object_store_0.10.0)
+
+**Breaking changes:**
+
+- Add put\_multipart\_opts \(\#5435\) [\#5652](https://github.com/apache/arrow-rs/pull/5652) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Add Attributes API \(\#5329\) [\#5650](https://github.com/apache/arrow-rs/pull/5650) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Support non-contiguous put payloads / vectored writes \(\#5514\) [\#5538](https://github.com/apache/arrow-rs/pull/5538) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Replace AsyncWrite with Upload trait and rename MultiPartStore to MultipartStore \(\#5458\) [\#5500](https://github.com/apache/arrow-rs/pull/5500) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
 
 **Implemented enhancements:**
 
-- \[object\_store\] Enable anonymous read access for Azure [\#5424](https://github.com/apache/arrow-rs/issues/5424) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Support for additional URL formats in object\_store for Azure blob [\#5370](https://github.com/apache/arrow-rs/issues/5370) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Mention "Http" support in README [\#5320](https://github.com/apache/arrow-rs/issues/5320) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Pass Options to HttpBuilder in parse\_url\_opts [\#5310](https://github.com/apache/arrow-rs/issues/5310) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Remove Localstack DynamoDB Workaround Once Fixed Upstream [\#5267](https://github.com/apache/arrow-rs/issues/5267) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Can I use S3 server side encryption [\#5087](https://github.com/apache/arrow-rs/issues/5087) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Improve Retry Coverage [\#5608](https://github.com/apache/arrow-rs/issues/5608) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Zero Copy Support [\#5593](https://github.com/apache/arrow-rs/issues/5593)
+- ObjectStore bulk delete [\#5591](https://github.com/apache/arrow-rs/issues/5591)
+- Retry on Broken Connection [\#5589](https://github.com/apache/arrow-rs/issues/5589)
+- Inconsistent Multipart Nomenclature [\#5526](https://github.com/apache/arrow-rs/issues/5526) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- \[ObjectStore\] Non-Contiguous Write Payloads [\#5514](https://github.com/apache/arrow-rs/issues/5514) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- In Object Store, return version & etag on multipart put. [\#5443](https://github.com/apache/arrow-rs/issues/5443) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- Release Object Store 0.9.1 [\#5436](https://github.com/apache/arrow-rs/issues/5436) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- object\_store: allow setting content-type per request [\#5329](https://github.com/apache/arrow-rs/issues/5329) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- GCS Signed URL Support [\#5233](https://github.com/apache/arrow-rs/issues/5233)
 
 **Fixed bugs:**
 
-- delete\_stream fails in MinIO [\#5414](https://github.com/apache/arrow-rs/issues/5414) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- \[object\_store\] Completing an empty Multipart Upload fails for AWS S3 [\#5404](https://github.com/apache/arrow-rs/issues/5404) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Multipart upload can leave futures unpolled, leading to timeout [\#5366](https://github.com/apache/arrow-rs/issues/5366) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
-- Broken Link in README \(Rust Object Store\) Content [\#5309](https://github.com/apache/arrow-rs/issues/5309) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- \[object\_store\] minor bug: typos present in local variable  [\#5628](https://github.com/apache/arrow-rs/issues/5628) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- \[arrow-csv\] Schema inference requires csv on disk [\#5551](https://github.com/apache/arrow-rs/issues/5551)
+- Local object store copy/rename with nonexistent `from` file loops forever instead of erroring [\#5503](https://github.com/apache/arrow-rs/issues/5503) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- object store ApplicationDefaultCredentials auth is not working on windows  [\#5466](https://github.com/apache/arrow-rs/issues/5466) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+- MicrosoftAzure store list result omits empty objects [\#5451](https://github.com/apache/arrow-rs/issues/5451) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)]
+
+**Documentation updates:**
+
+- Minor: add additional documentation about `BufWriter` [\#5519](https://github.com/apache/arrow-rs/pull/5519) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([alamb](https://github.com/alamb))
 
 **Merged pull requests:**
 
-- Expose path\_to\_filesystem public [\#5441](https://github.com/apache/arrow-rs/pull/5441) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([metesynnada](https://github.com/metesynnada))
-- Update nix requirement from 0.27.1 to 0.28.0 in /object\_store [\#5432](https://github.com/apache/arrow-rs/pull/5432) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([dependabot[bot]](https://github.com/apps/dependabot))
-- Add BufWriter for Adapative Put / Multipart Upload [\#5431](https://github.com/apache/arrow-rs/pull/5431) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
-- Enable anonymous access for MicrosoftAzure [\#5425](https://github.com/apache/arrow-rs/pull/5425) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([andrebsguedes](https://github.com/andrebsguedes))
-- fix\(object\_store\): Include Content-MD5 header for S3 DeleteObjects [\#5415](https://github.com/apache/arrow-rs/pull/5415) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([paraseba](https://github.com/paraseba))
-- docds\(object\_store\): Mention HTTP/WebDAV in README [\#5409](https://github.com/apache/arrow-rs/pull/5409) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([Xuanwo](https://github.com/Xuanwo))
-- \[object\_store\] Fix empty Multipart Upload for AWS S3 [\#5405](https://github.com/apache/arrow-rs/pull/5405) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([andrebsguedes](https://github.com/andrebsguedes))
-- feat: S3 server-side encryption [\#5402](https://github.com/apache/arrow-rs/pull/5402) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([wjones127](https://github.com/wjones127))
-- Pull container name from URL for Azure blob [\#5371](https://github.com/apache/arrow-rs/pull/5371) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([bradvoth](https://github.com/bradvoth))
-- docs\(object-store\): add warning to flush [\#5369](https://github.com/apache/arrow-rs/pull/5369) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([wjones127](https://github.com/wjones127))
-- Minor\(docs\): update master to main for DataFusion/Ballista [\#5363](https://github.com/apache/arrow-rs/pull/5363) ([caicancai](https://github.com/caicancai))
-- Test parse\_url\_opts for HTTP \(\#5310\) [\#5316](https://github.com/apache/arrow-rs/pull/5316) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
-- Update IOx links [\#5312](https://github.com/apache/arrow-rs/pull/5312) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
-- Pass options to HTTPBuilder in parse\_url\_opts \(\#5310\) [\#5311](https://github.com/apache/arrow-rs/pull/5311) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
-- Bump actions/cache from 3 to 4 [\#5308](https://github.com/apache/arrow-rs/pull/5308) ([dependabot[bot]](https://github.com/apps/dependabot))
-- Remove localstack DynamoDB workaround \(\#5267\) [\#5307](https://github.com/apache/arrow-rs/pull/5307) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
-- refactor: log server error during object store retries [\#5294](https://github.com/apache/arrow-rs/pull/5294) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([crepererum](https://github.com/crepererum))
-- Prepare arrow 50.0.0 [\#5291](https://github.com/apache/arrow-rs/pull/5291) ([tustvold](https://github.com/tustvold))
-- Enable JS tests again [\#5287](https://github.com/apache/arrow-rs/pull/5287) ([domoritz](https://github.com/domoritz))
+- minor-fix: removed typos in object\_store sub crate [\#5629](https://github.com/apache/arrow-rs/pull/5629) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([Silemo](https://github.com/Silemo))
+- Retry on More Error Classes [\#5609](https://github.com/apache/arrow-rs/pull/5609) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([andrebsguedes](https://github.com/andrebsguedes))
+- Fix handling of empty multipart uploads for GCS [\#5590](https://github.com/apache/arrow-rs/pull/5590) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Upgrade object\_store dependency to use chrono `0.4.34` [\#5578](https://github.com/apache/arrow-rs/pull/5578) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([l1nxy](https://github.com/l1nxy))
+- Fix Latest Clippy Lints for object\_store [\#5546](https://github.com/apache/arrow-rs/pull/5546) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Update reqwest 0.12 and http 1.0 [\#5536](https://github.com/apache/arrow-rs/pull/5536) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Implement MultipartStore for ThrottledStore [\#5533](https://github.com/apache/arrow-rs/pull/5533) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- fix: copy/rename return error if source is nonexistent [\#5528](https://github.com/apache/arrow-rs/pull/5528) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([dimbtp](https://github.com/dimbtp))
+- Prepare arrow 51.0.0 [\#5516](https://github.com/apache/arrow-rs/pull/5516) ([tustvold](https://github.com/tustvold))
+- Implement MultiPartStore for InMemory [\#5495](https://github.com/apache/arrow-rs/pull/5495) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([tustvold](https://github.com/tustvold))
+- Add more comprehensive documentation on testing and benchmarking to CONTRIBUTING.md [\#5478](https://github.com/apache/arrow-rs/pull/5478) ([monkwire](https://github.com/monkwire))
+- add support for gcp application default auth on windows in object store [\#5473](https://github.com/apache/arrow-rs/pull/5473) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([Itayazolay](https://github.com/Itayazolay))
+- Update base64 requirement from 0.21 to 0.22 in /object\_store [\#5465](https://github.com/apache/arrow-rs/pull/5465) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([dependabot[bot]](https://github.com/apps/dependabot))
+- Uses ResourceType for filtering list directories instead of workaround [\#5452](https://github.com/apache/arrow-rs/pull/5452) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([andrebsguedes](https://github.com/andrebsguedes))
+- Add GCS signed URL support [\#5300](https://github.com/apache/arrow-rs/pull/5300) [[object-store](https://github.com/apache/arrow-rs/labels/object-store)] ([l1nxy](https://github.com/l1nxy))
 
 \* *This Changelog was automatically generated by [github_changelog_generator](https://github.com/github-changelog-generator/github-changelog-generator)*
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index d0c3af240..ca1e5b327 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -17,7 +17,7 @@
 
 [package]
 name = "object_store"
-version = "0.9.1"
+version = "0.10.0"
 edition = "2021"
 license = "MIT/Apache-2.0"
 readme = "README.md"
diff --git a/object_store/dev/release/update_change_log.sh b/object_store/dev/release/update_change_log.sh
index 83d5c323c..5a28409e3 100755
--- a/object_store/dev/release/update_change_log.sh
+++ b/object_store/dev/release/update_change_log.sh
@@ -29,8 +29,8 @@
 
 set -e
 
-SINCE_TAG="object_store_0.9.0"
-FUTURE_RELEASE="object_store_0.9.1"
+SINCE_TAG="object_store_0.9.1"
+FUTURE_RELEASE="object_store_0.10.0"
 
 SOURCE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 SOURCE_TOP_DIR="$(cd "${SOURCE_DIR}/../../" && pwd)"
diff --git a/object_store/src/attributes.rs b/object_store/src/attributes.rs
new file mode 100644
index 000000000..9b90b5325
--- /dev/null
+++ b/object_store/src/attributes.rs
@@ -0,0 +1,211 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::borrow::Cow;
+use std::collections::HashMap;
+use std::ops::Deref;
+
+/// Additional object attribute types
+#[non_exhaustive]
+#[derive(Debug, Hash, Eq, PartialEq, Clone)]
+pub enum Attribute {
+    /// Specifies the MIME type of the object
+    ///
+    /// This takes precedence over any [ClientOptions](crate::ClientOptions) configuration
+    ///
+    /// See [Content-Type](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Type)
+    ContentType,
+    /// Overrides cache control policy of the object
+    ///
+    /// See [Cache-Control](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Cache-Control)
+    CacheControl,
+}
+
+/// The value of an [`Attribute`]
+///
+/// Provides efficient conversion from both static and owned strings
+///
+/// ```
+/// # use object_store::AttributeValue;
+/// // Can use static strings without needing an allocation
+/// let value = AttributeValue::from("bar");
+/// // Can also store owned strings
+/// let value = AttributeValue::from("foo".to_string());
+/// ```
+#[derive(Debug, Hash, Eq, PartialEq, Clone)]
+pub struct AttributeValue(Cow<'static, str>);
+
+impl AsRef<str> for AttributeValue {
+    fn as_ref(&self) -> &str {
+        &self.0
+    }
+}
+
+impl From<&'static str> for AttributeValue {
+    fn from(value: &'static str) -> Self {
+        Self(Cow::Borrowed(value))
+    }
+}
+
+impl From<String> for AttributeValue {
+    fn from(value: String) -> Self {
+        Self(Cow::Owned(value))
+    }
+}
+
+impl Deref for AttributeValue {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        self.0.as_ref()
+    }
+}
+
+/// Additional attributes of an object
+///
+/// Attributes can be specified in [PutOptions](crate::PutOptions) and retrieved
+/// from APIs returning [GetResult](crate::GetResult).
+///
+/// Unlike [`ObjectMeta`](crate::ObjectMeta), [`Attributes`] are not returned by
+/// listing APIs
+#[derive(Debug, Default, Eq, PartialEq, Clone)]
+pub struct Attributes(HashMap<Attribute, AttributeValue>);
+
+impl Attributes {
+    /// Create a new empty [`Attributes`]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Create a new [`Attributes`] with space for `capacity` [`Attribute`]
+    pub fn with_capacity(capacity: usize) -> Self {
+        Self(HashMap::with_capacity(capacity))
+    }
+
+    /// Insert a new [`Attribute`], [`AttributeValue`] pair
+    ///
+    /// Returns the previous value for `key` if any
+    pub fn insert(&mut self, key: Attribute, value: AttributeValue) -> Option<AttributeValue> {
+        self.0.insert(key, value)
+    }
+
+    /// Returns the [`AttributeValue`] for `key` if any
+    pub fn get(&self, key: &Attribute) -> Option<&AttributeValue> {
+        self.0.get(key)
+    }
+
+    /// Removes the [`AttributeValue`] for `key` if any
+    pub fn remove(&mut self, key: &Attribute) -> Option<AttributeValue> {
+        self.0.remove(key)
+    }
+
+    /// Returns an [`AttributesIter`] over this
+    pub fn iter(&self) -> AttributesIter<'_> {
+        self.into_iter()
+    }
+
+    /// Returns the number of [`Attribute`] in this collection
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    /// Returns true if this contains no [`Attribute`]
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+}
+
+impl<K, V> FromIterator<(K, V)> for Attributes
+where
+    K: Into<Attribute>,
+    V: Into<AttributeValue>,
+{
+    fn from_iter<T: IntoIterator<Item = (K, V)>>(iter: T) -> Self {
+        Self(
+            iter.into_iter()
+                .map(|(k, v)| (k.into(), v.into()))
+                .collect(),
+        )
+    }
+}
+
+impl<'a> IntoIterator for &'a Attributes {
+    type Item = (&'a Attribute, &'a AttributeValue);
+    type IntoIter = AttributesIter<'a>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        AttributesIter(self.0.iter())
+    }
+}
+
+/// Iterator over [`Attributes`]
+#[derive(Debug)]
+pub struct AttributesIter<'a>(std::collections::hash_map::Iter<'a, Attribute, AttributeValue>);
+
+impl<'a> Iterator for AttributesIter<'a> {
+    type Item = (&'a Attribute, &'a AttributeValue);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.0.next()
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.0.size_hint()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_attributes_basic() {
+        let mut attributes = Attributes::from_iter([
+            (Attribute::ContentType, "test"),
+            (Attribute::CacheControl, "control"),
+        ]);
+
+        assert!(!attributes.is_empty());
+        assert_eq!(attributes.len(), 2);
+
+        assert_eq!(
+            attributes.get(&Attribute::ContentType),
+            Some(&"test".into())
+        );
+
+        let metav = "control".into();
+        assert_eq!(attributes.get(&Attribute::CacheControl), Some(&metav));
+        assert_eq!(
+            attributes.insert(Attribute::CacheControl, "v1".into()),
+            Some(metav)
+        );
+        assert_eq!(attributes.len(), 2);
+
+        assert_eq!(
+            attributes.remove(&Attribute::CacheControl).unwrap(),
+            "v1".into()
+        );
+        assert_eq!(attributes.len(), 1);
+
+        let metav: AttributeValue = "v2".into();
+        attributes.insert(Attribute::CacheControl, metav.clone());
+        assert_eq!(attributes.get(&Attribute::CacheControl), Some(&metav));
+        assert_eq!(attributes.len(), 2);
+    }
+}
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index c1789ed14..4a4dc178d 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -19,8 +19,8 @@ use crate::aws::builder::S3EncryptionHeaders;
 use crate::aws::checksum::Checksum;
 use crate::aws::credential::{AwsCredential, CredentialExt};
 use crate::aws::{
-    AwsAuthorizer, AwsCredentialProvider, S3ConditionalPut, S3CopyIfNotExists, STORE,
-    STRICT_PATH_ENCODE_SET,
+    AwsAuthorizer, AwsCredentialProvider, S3ConditionalPut, S3CopyIfNotExists, COPY_SOURCE_HEADER,
+    STORE, STRICT_PATH_ENCODE_SET, TAGS_HEADER,
 };
 use crate::client::get::GetClient;
 use crate::client::header::{get_etag, HeaderConfig};
@@ -35,23 +35,21 @@ use crate::client::GetOptionsExt;
 use crate::multipart::PartId;
 use crate::path::DELIMITER;
 use crate::{
-    ClientOptions, GetOptions, ListResult, MultipartId, Path, PutPayload, PutResult, Result,
-    RetryConfig,
+    Attribute, Attributes, ClientOptions, GetOptions, ListResult, MultipartId, Path,
+    PutMultipartOpts, PutPayload, PutResult, Result, RetryConfig, TagSet,
 };
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::{Buf, Bytes};
-use hyper::http;
+use hyper::header::{CACHE_CONTROL, CONTENT_LENGTH};
 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_LENGTH, CONTENT_TYPE},
-    Client as ReqwestClient, Method, RequestBuilder, Response,
-};
+use reqwest::{header::CONTENT_TYPE, Client as ReqwestClient, Method, RequestBuilder, Response};
 use ring::digest;
 use ring::digest::Context;
 use serde::{Deserialize, Serialize};
@@ -100,9 +98,6 @@ pub(crate) enum Error {
     #[snafu(display("Error getting list response body: {}", source))]
     ListResponseBody { source: reqwest::Error },
 
-    #[snafu(display("Error performing create multipart request: {}", source))]
-    CreateMultipartRequest { source: crate::client::retry::Error },
-
     #[snafu(display("Error getting create multipart response body: {}", source))]
     CreateMultipartResponseBody { source: reqwest::Error },
 
@@ -291,8 +286,75 @@ impl<'a> Request<'a> {
         Self { builder, ..self }
     }
 
-    pub fn idempotent(mut self, idempotent: bool) -> Self {
-        self.idempotent = idempotent;
+    pub fn headers(self, headers: HeaderMap) -> Self {
+        let builder = self.builder.headers(headers);
+        Self { builder, ..self }
+    }
+
+    pub fn idempotent(self, idempotent: bool) -> Self {
+        Self { idempotent, ..self }
+    }
+
+    pub fn with_encryption_headers(self) -> Self {
+        let headers = self.config.encryption_headers.clone().into();
+        let builder = self.builder.headers(headers);
+        Self { builder, ..self }
+    }
+
+    pub fn with_session_creds(self, use_session_creds: bool) -> Self {
+        Self {
+            use_session_creds,
+            ..self
+        }
+    }
+
+    pub fn with_tags(mut self, tags: TagSet) -> Self {
+        let tags = tags.encoded();
+        if !tags.is_empty() && !self.config.disable_tagging {
+            self.builder = self.builder.header(&TAGS_HEADER, tags);
+        }
+        self
+    }
+
+    pub fn with_attributes(self, attributes: Attributes) -> Self {
+        let mut has_content_type = false;
+        let mut builder = self.builder;
+        for (k, v) in &attributes {
+            builder = match k {
+                Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()),
+                Attribute::ContentType => {
+                    has_content_type = true;
+                    builder.header(CONTENT_TYPE, v.as_ref())
+                }
+            };
+        }
+
+        if !has_content_type {
+            if let Some(value) = self.config.client_options.get_content_type(self.path) {
+                builder = builder.header(CONTENT_TYPE, value);
+            }
+        }
+        Self { builder, ..self }
+    }
+
+    pub fn with_payload(mut self, payload: PutPayload) -> Self {
+        if !self.config.skip_signature || self.config.checksum.is_some() {
+            let mut sha256 = Context::new(&digest::SHA256);
+            payload.iter().for_each(|x| sha256.update(x));
+            let payload_sha256 = sha256.finish();
+
+            if let Some(Checksum::SHA256) = self.config.checksum {
+                self.builder = self.builder.header(
+                    "x-amz-checksum-sha256",
+                    BASE64_STANDARD.encode(payload_sha256),
+                );
+            }
+            self.payload_sha256 = Some(payload_sha256);
+        }
+
+        let content_length = payload.content_length();
+        self.builder = self.builder.header(CONTENT_LENGTH, content_length);
+        self.payload = Some(payload);
         self
     }
 
@@ -337,67 +399,19 @@ impl S3Client {
         Ok(Self { config, client })
     }
 
-    /// Make an S3 PUT request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html>
-    ///
-    /// Returns the ETag
-    pub fn put_request<'a>(
-        &'a self,
-        path: &'a Path,
-        payload: PutPayload,
-        with_encryption_headers: bool,
-    ) -> Request<'a> {
+    pub fn request<'a>(&'a self, method: Method, path: &'a Path) -> Request<'a> {
         let url = self.config.path_url(path);
-        let mut builder = self.client.request(Method::PUT, url);
-        if with_encryption_headers {
-            builder = builder.headers(self.config.encryption_headers.clone().into());
-        }
-
-        let mut sha256 = Context::new(&digest::SHA256);
-        payload.iter().for_each(|x| sha256.update(x));
-        let payload_sha256 = sha256.finish();
-
-        if let Some(Checksum::SHA256) = self.config.checksum {
-            builder = builder.header(
-                "x-amz-checksum-sha256",
-                BASE64_STANDARD.encode(payload_sha256),
-            )
-        }
-
-        if let Some(value) = self.config.client_options.get_content_type(path) {
-            builder = builder.header(CONTENT_TYPE, value);
-        }
-
         Request {
             path,
-            builder: builder.header(CONTENT_LENGTH, payload.content_length()),
-            payload: Some(payload),
-            payload_sha256: Some(payload_sha256),
+            builder: self.client.request(method, url),
+            payload: None,
+            payload_sha256: None,
             config: &self.config,
             use_session_creds: true,
             idempotent: false,
         }
     }
 
-    /// Make an S3 Delete request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObject.html>
-    pub async fn delete_request<T: Serialize + ?Sized + Sync>(
-        &self,
-        path: &Path,
-        query: &T,
-    ) -> Result<()> {
-        let credential = self.config.get_session_credential().await?;
-        let url = self.config.path_url(path);
-
-        self.client
-            .request(Method::DELETE, url)
-            .query(query)
-            .with_aws_sigv4(credential.authorizer(), None)
-            .send_retry(&self.config.retry_config)
-            .await
-            .map_err(|e| e.error(STORE, path.to_string()))?;
-
-        Ok(())
-    }
-
     /// Make an S3 Delete Objects request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html>
     ///
     /// Produces a vector of results, one for each path in the input vector. If
@@ -501,41 +515,29 @@ impl S3Client {
     }
 
     /// Make an S3 Copy request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html>
-    pub fn copy_request<'a>(&'a self, from: &'a Path, to: &Path) -> Request<'a> {
-        let url = self.config.path_url(to);
+    pub fn copy_request<'a>(&'a self, from: &Path, to: &'a Path) -> Request<'a> {
         let source = format!("{}/{}", self.config.bucket, encode_path(from));
-
-        let builder = self
-            .client
-            .request(Method::PUT, url)
-            .header("x-amz-copy-source", source)
-            .headers(self.config.encryption_headers.clone().into());
-
-        Request {
-            builder,
-            path: from,
-            config: &self.config,
-            payload: None,
-            payload_sha256: None,
-            use_session_creds: false,
-            idempotent: false,
-        }
+        self.request(Method::PUT, to)
+            .idempotent(true)
+            .header(&COPY_SOURCE_HEADER, &source)
+            .headers(self.config.encryption_headers.clone().into())
+            .with_session_creds(false)
     }
 
-    pub async fn create_multipart(&self, location: &Path) -> Result<MultipartId> {
-        let credential = self.config.get_session_credential().await?;
-        let url = format!("{}?uploads=", self.config.path_url(location),);
-
+    pub async fn create_multipart(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<MultipartId> {
         let response = self
-            .client
-            .request(Method::POST, url)
-            .headers(self.config.encryption_headers.clone().into())
-            .with_aws_sigv4(credential.authorizer(), None)
-            .retryable(&self.config.retry_config)
+            .request(Method::POST, location)
+            .query(&[("uploads", "")])
+            .with_encryption_headers()
+            .with_attributes(opts.attributes)
+            .with_tags(opts.tags)
             .idempotent(true)
             .send()
-            .await
-            .context(CreateMultipartRequestSnafu)?
+            .await?
             .bytes()
             .await
             .context(CreateMultipartResponseBodySnafu)?;
@@ -556,7 +558,8 @@ impl S3Client {
         let part = (part_idx + 1).to_string();
 
         let response = self
-            .put_request(path, data, false)
+            .request(Method::PUT, path)
+            .with_payload(data)
             .query(&[("partNumber", &part), ("uploadId", upload_id)])
             .idempotent(true)
             .send()
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 9e741c914..7f1edf12f 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -45,10 +45,12 @@ use crate::signer::Signer;
 use crate::util::STRICT_ENCODE_SET;
 use crate::{
     Error, GetOptions, GetResult, ListResult, MultipartId, MultipartUpload, ObjectMeta,
-    ObjectStore, Path, PutMode, PutOptions, PutPayload, PutResult, Result, UploadPart,
+    ObjectStore, Path, PutMode, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
+    UploadPart,
 };
 
 static TAGS_HEADER: HeaderName = HeaderName::from_static("x-amz-tagging");
+static COPY_SOURCE_HEADER: HeaderName = HeaderName::from_static("x-amz-copy-source");
 
 mod builder;
 mod checksum;
@@ -156,11 +158,13 @@ impl ObjectStore for AmazonS3 {
         payload: PutPayload,
         opts: PutOptions,
     ) -> Result<PutResult> {
-        let mut request = self.client.put_request(location, payload, true);
-        let tags = opts.tags.encoded();
-        if !tags.is_empty() && !self.client.config.disable_tagging {
-            request = request.header(&TAGS_HEADER, tags);
-        }
+        let request = self
+            .client
+            .request(Method::PUT, location)
+            .with_payload(payload)
+            .with_attributes(opts.attributes)
+            .with_tags(opts.tags)
+            .with_encryption_headers();
 
         match (opts.mode, &self.client.config.conditional_put) {
             (PutMode::Overwrite, _) => request.idempotent(true).do_put().await,
@@ -203,8 +207,12 @@ impl ObjectStore for AmazonS3 {
         }
     }
 
-    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
-        let upload_id = self.client.create_multipart(location).await?;
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        let upload_id = self.client.create_multipart(location, opts).await?;
 
         Ok(Box::new(S3MultiPartUpload {
             part_idx: 0,
@@ -222,7 +230,8 @@ impl ObjectStore for AmazonS3 {
     }
 
     async fn delete(&self, location: &Path) -> Result<()> {
-        self.client.delete_request(location, &()).await
+        self.client.request(Method::DELETE, location).send().await?;
+        Ok(())
     }
 
     fn delete_stream<'a>(
@@ -350,15 +359,22 @@ impl MultipartUpload for S3MultiPartUpload {
     async fn abort(&mut self) -> Result<()> {
         self.state
             .client
-            .delete_request(&self.state.location, &[("uploadId", &self.state.upload_id)])
-            .await
+            .request(Method::DELETE, &self.state.location)
+            .query(&[("uploadId", &self.state.upload_id)])
+            .idempotent(true)
+            .send()
+            .await?;
+
+        Ok(())
     }
 }
 
 #[async_trait]
 impl MultipartStore for AmazonS3 {
     async fn create_multipart(&self, path: &Path) -> Result<MultipartId> {
-        self.client.create_multipart(path).await
+        self.client
+            .create_multipart(path, PutMultipartOpts::default())
+            .await
     }
 
     async fn put_part(
@@ -381,7 +397,12 @@ impl MultipartStore for AmazonS3 {
     }
 
     async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()> {
-        self.client.delete_request(path, &[("uploadId", id)]).await
+        self.client
+            .request(Method::DELETE, path)
+            .query(&[("uploadId", id)])
+            .send()
+            .await?;
+        Ok(())
     }
 }
 
@@ -403,7 +424,7 @@ mod tests {
         let test_not_exists = config.copy_if_not_exists.is_some();
         let test_conditional_put = config.conditional_put.is_some();
 
-        put_get_delete_list_opts(&integration).await;
+        put_get_delete_list(&integration).await;
         get_opts(&integration).await;
         list_uses_directories_correctly(&integration).await;
         list_with_delimiter(&integration).await;
@@ -412,6 +433,7 @@ mod tests {
         multipart(&integration, &integration).await;
         signing(&integration).await;
         s3_encryption(&integration).await;
+        put_get_attributes(&integration).await;
 
         // Object tagging is not supported by S3 Express One Zone
         if config.session_provider.is_none() {
@@ -432,12 +454,12 @@ mod tests {
         // run integration test with unsigned payload enabled
         let builder = AmazonS3Builder::from_env().with_unsigned_payload(true);
         let integration = builder.build().unwrap();
-        put_get_delete_list_opts(&integration).await;
+        put_get_delete_list(&integration).await;
 
         // run integration test with checksum set to sha256
         let builder = AmazonS3Builder::from_env().with_checksum_algorithm(Checksum::SHA256);
         let integration = builder.build().unwrap();
-        put_get_delete_list_opts(&integration).await;
+        put_get_delete_list(&integration).await;
 
         match &integration.client.config.copy_if_not_exists {
             Some(S3CopyIfNotExists::Dynamo(d)) => dynamo::integration_test(&integration, d).await,
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index d5972d0a8..918fcd047 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -27,8 +27,8 @@ use crate::multipart::PartId;
 use crate::path::DELIMITER;
 use crate::util::{deserialize_rfc1123, GetRange};
 use crate::{
-    ClientOptions, GetOptions, ListResult, ObjectMeta, Path, PutMode, PutOptions, PutPayload,
-    PutResult, Result, RetryConfig,
+    Attribute, Attributes, ClientOptions, GetOptions, ListResult, ObjectMeta, Path, PutMode,
+    PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, RetryConfig, TagSet,
 };
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
@@ -36,7 +36,6 @@ use base64::Engine;
 use bytes::{Buf, Bytes};
 use chrono::{DateTime, Utc};
 use hyper::http::HeaderName;
-use reqwest::header::CONTENT_TYPE;
 use reqwest::{
     header::{HeaderValue, CONTENT_LENGTH, IF_MATCH, IF_NONE_MATCH},
     Client as ReqwestClient, Method, RequestBuilder, Response,
@@ -49,6 +48,8 @@ use std::time::Duration;
 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 TAGS_HEADER: HeaderName = HeaderName::from_static("x-ms-tags");
 
@@ -187,11 +188,39 @@ impl<'a> PutRequest<'a> {
         Self { builder, ..self }
     }
 
-    fn set_idempotent(mut self, idempotent: bool) -> Self {
-        self.idempotent = idempotent;
+    fn idempotent(self, idempotent: bool) -> Self {
+        Self { idempotent, ..self }
+    }
+
+    fn with_tags(mut self, tags: TagSet) -> Self {
+        let tags = tags.encoded();
+        if !tags.is_empty() && !self.config.disable_tagging {
+            self.builder = self.builder.header(&TAGS_HEADER, tags);
+        }
         self
     }
 
+    fn with_attributes(self, attributes: Attributes) -> Self {
+        let mut builder = self.builder;
+        let mut has_content_type = false;
+        for (k, v) in &attributes {
+            builder = match k {
+                Attribute::CacheControl => builder.header(&MS_CACHE_CONTROL, v.as_ref()),
+                Attribute::ContentType => {
+                    has_content_type = true;
+                    builder.header(&MS_CONTENT_TYPE, v.as_ref())
+                }
+            };
+        }
+
+        if !has_content_type {
+            if let Some(value) = self.config.client_options.get_content_type(self.path) {
+                builder = builder.header(&MS_CONTENT_TYPE, value);
+            }
+        }
+        Self { builder, ..self }
+    }
+
     async fn send(self) -> Result<Response> {
         let credential = self.config.get_credential().await?;
         let response = self
@@ -199,7 +228,7 @@ impl<'a> PutRequest<'a> {
             .header(CONTENT_LENGTH, self.payload.content_length())
             .with_azure_authorization(&credential, &self.config.account)
             .retryable(&self.config.retry_config)
-            .idempotent(true)
+            .idempotent(self.idempotent)
             .payload(Some(self.payload))
             .send()
             .await
@@ -235,12 +264,7 @@ impl AzureClient {
 
     fn put_request<'a>(&'a self, path: &'a Path, payload: PutPayload) -> PutRequest<'a> {
         let url = self.config.path_url(path);
-
-        let mut builder = self.client.request(Method::PUT, url);
-
-        if let Some(value) = self.config().client_options.get_content_type(path) {
-            builder = builder.header(CONTENT_TYPE, value);
-        }
+        let builder = self.client.request(Method::PUT, url);
 
         PutRequest {
             path,
@@ -258,10 +282,13 @@ impl AzureClient {
         payload: PutPayload,
         opts: PutOptions,
     ) -> Result<PutResult> {
-        let builder = self.put_request(path, payload);
+        let builder = self
+            .put_request(path, payload)
+            .with_attributes(opts.attributes)
+            .with_tags(opts.tags);
 
         let builder = match &opts.mode {
-            PutMode::Overwrite => builder.set_idempotent(true),
+            PutMode::Overwrite => builder.idempotent(true),
             PutMode::Create => builder.header(&IF_NONE_MATCH, "*"),
             PutMode::Update(v) => {
                 let etag = v.e_tag.as_ref().context(MissingETagSnafu)?;
@@ -269,11 +296,6 @@ impl AzureClient {
             }
         };
 
-        let builder = match (opts.tags.encoded(), self.config.disable_tagging) {
-            ("", _) | (_, true) => builder,
-            (tags, false) => builder.header(&TAGS_HEADER, tags),
-        };
-
         let response = builder.header(&BLOB_TYPE, "BlockBlob").send().await?;
         Ok(get_put_result(response.headers(), VERSION_HEADER).context(MetadataSnafu)?)
     }
@@ -290,7 +312,7 @@ impl AzureClient {
 
         self.put_request(path, payload)
             .query(&[("comp", "block"), ("blockid", &block_id)])
-            .set_idempotent(true)
+            .idempotent(true)
             .send()
             .await?;
 
@@ -298,16 +320,24 @@ impl AzureClient {
     }
 
     /// PUT a block list <https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-list>
-    pub async fn put_block_list(&self, path: &Path, parts: Vec<PartId>) -> Result<PutResult> {
+    pub async fn put_block_list(
+        &self,
+        path: &Path,
+        parts: Vec<PartId>,
+        opts: PutMultipartOpts,
+    ) -> Result<PutResult> {
         let blocks = parts
             .into_iter()
             .map(|part| BlockId::from(part.content_id))
             .collect();
 
+        let payload = BlockList { blocks }.to_xml().into();
         let response = self
-            .put_request(path, BlockList { blocks }.to_xml().into())
+            .put_request(path, payload)
+            .with_attributes(opts.attributes)
+            .with_tags(opts.tags)
             .query(&[("comp", "blocklist")])
-            .set_idempotent(true)
+            .idempotent(true)
             .send()
             .await?;
 
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index 8dc52422b..25ae6dda6 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -27,7 +27,7 @@ use crate::{
     path::Path,
     signer::Signer,
     GetOptions, GetResult, ListResult, MultipartId, MultipartUpload, ObjectMeta, ObjectStore,
-    PutOptions, PutPayload, PutResult, Result, UploadPart,
+    PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, UploadPart,
 };
 use async_trait::async_trait;
 use futures::stream::BoxStream;
@@ -95,9 +95,14 @@ impl ObjectStore for MicrosoftAzure {
         self.client.put_blob(location, payload, opts).await
     }
 
-    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
         Ok(Box::new(AzureMultiPartUpload {
             part_idx: 0,
+            opts,
             state: Arc::new(UploadState {
                 client: Arc::clone(&self.client),
                 location: location.clone(),
@@ -196,6 +201,7 @@ impl Signer for MicrosoftAzure {
 struct AzureMultiPartUpload {
     part_idx: usize,
     state: Arc<UploadState>,
+    opts: PutMultipartOpts,
 }
 
 #[derive(Debug)]
@@ -223,7 +229,7 @@ impl MultipartUpload for AzureMultiPartUpload {
 
         self.state
             .client
-            .put_block_list(&self.state.location, parts)
+            .put_block_list(&self.state.location, parts, std::mem::take(&mut self.opts))
             .await
     }
 
@@ -255,7 +261,9 @@ impl MultipartStore for MicrosoftAzure {
         _: &MultipartId,
         parts: Vec<PartId>,
     ) -> Result<PutResult> {
-        self.client.put_block_list(path, parts).await
+        self.client
+            .put_block_list(path, parts, Default::default())
+            .await
     }
 
     async fn abort_multipart(&self, _: &Path, _: &MultipartId) -> Result<()> {
@@ -276,7 +284,7 @@ mod tests {
         crate::test_util::maybe_skip_integration!();
         let integration = MicrosoftAzureBuilder::from_env().build().unwrap();
 
-        put_get_delete_list_opts(&integration).await;
+        put_get_delete_list(&integration).await;
         get_opts(&integration).await;
         list_uses_directories_correctly(&integration).await;
         list_with_delimiter(&integration).await;
@@ -292,7 +300,12 @@ mod tests {
             let client = Arc::clone(&integration.client);
             async move { client.get_blob_tagging(&p).await }
         })
-        .await
+        .await;
+
+        // Azurite doesn't support attributes properly
+        if !integration.client.config().is_emulator {
+            put_get_attributes(&integration).await;
+        }
     }
 
     #[ignore = "Used for manual testing against a real storage account."]
diff --git a/object_store/src/chunked.rs b/object_store/src/chunked.rs
index 9abe49dbf..a3bd76267 100644
--- a/object_store/src/chunked.rs
+++ b/object_store/src/chunked.rs
@@ -29,7 +29,7 @@ use futures::StreamExt;
 use crate::path::Path;
 use crate::{
     GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
-    PutOptions, PutResult,
+    PutMultipartOpts, PutOptions, PutResult,
 };
 use crate::{PutPayload, Result};
 
@@ -75,6 +75,14 @@ impl ObjectStore for ChunkedStore {
         self.inner.put_multipart(location).await
     }
 
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        self.inner.put_multipart_opts(location, opts).await
+    }
+
     async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
         let r = self.inner.get_opts(location, options).await?;
         let stream = match r.payload {
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 2e399e523..f70045761 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -19,10 +19,10 @@ use std::ops::Range;
 
 use crate::client::header::{header_meta, HeaderConfig};
 use crate::path::Path;
-use crate::{GetOptions, GetRange, GetResult, GetResultPayload, Result};
+use crate::{Attribute, Attributes, GetOptions, GetRange, GetResult, GetResultPayload, Result};
 use async_trait::async_trait;
 use futures::{StreamExt, TryStreamExt};
-use hyper::header::CONTENT_RANGE;
+use hyper::header::{CACHE_CONTROL, CONTENT_RANGE, CONTENT_TYPE};
 use hyper::StatusCode;
 use reqwest::header::ToStrError;
 use reqwest::Response;
@@ -117,6 +117,12 @@ enum GetResultError {
     #[snafu(display("Content-Range header contained non UTF-8 characters"))]
     InvalidContentRange { source: ToStrError },
 
+    #[snafu(display("Cache-Control header contained non UTF-8 characters"))]
+    InvalidCacheControl { source: ToStrError },
+
+    #[snafu(display("Content-Type header contained non UTF-8 characters"))]
+    InvalidContentType { source: ToStrError },
+
     #[snafu(display("Requested {expected:?}, got {actual:?}"))]
     UnexpectedRange {
         expected: Range<usize>,
@@ -161,6 +167,16 @@ 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());
+    }
+
     let stream = response
         .bytes_stream()
         .map_err(|source| crate::Error::Generic {
@@ -172,6 +188,7 @@ fn get_result<T: GetClient>(
     Ok(GetResult {
         range,
         meta,
+        attributes,
         payload: GetResultPayload::Stream(stream),
     })
 }
diff --git a/object_store/src/client/mod.rs b/object_store/src/client/mod.rs
index 7728f3895..3fefbb568 100644
--- a/object_store/src/client/mod.rs
+++ b/object_store/src/client/mod.rs
@@ -485,7 +485,7 @@ impl ClientOptions {
     /// mime type if it was defined initially through
     /// `ClientOptions::with_content_type_for_suffix`
     ///
-    /// Otherwise returns the default mime type if it was defined
+    /// Otherwise, returns the default mime type if it was defined
     /// earlier through `ClientOptions::with_default_content_type`
     pub fn get_content_type(&self, path: &Path) -> Option<&str> {
         match path.extension() {
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index f91217f6f..9c39efe6b 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -29,22 +29,23 @@ use crate::multipart::PartId;
 use crate::path::{Path, DELIMITER};
 use crate::util::hex_encode;
 use crate::{
-    ClientOptions, GetOptions, ListResult, MultipartId, PutMode, PutOptions, PutPayload, PutResult,
-    Result, RetryConfig,
+    Attribute, Attributes, ClientOptions, GetOptions, ListResult, MultipartId, PutMode,
+    PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, RetryConfig,
 };
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::Buf;
-use hyper::header::CONTENT_LENGTH;
+use hyper::header::{CACHE_CONTROL, CONTENT_LENGTH, CONTENT_TYPE};
 use percent_encoding::{percent_encode, utf8_percent_encode, NON_ALPHANUMERIC};
 use reqwest::header::HeaderName;
-use reqwest::{header, Client, Method, RequestBuilder, Response, StatusCode};
+use reqwest::{Client, Method, RequestBuilder, Response, StatusCode};
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use std::sync::Arc;
 
 const VERSION_HEADER: &str = "x-goog-generation";
+const DEFAULT_CONTENT_TYPE: &str = "application/octet-stream";
 
 static VERSION_MATCH: HeaderName = HeaderName::from_static("x-goog-if-generation-match");
 
@@ -65,14 +66,8 @@ enum Error {
         path: String,
     },
 
-    #[snafu(display("Error performing delete request {}: {}", path, source))]
-    DeleteRequest {
-        source: crate::client::retry::Error,
-        path: String,
-    },
-
-    #[snafu(display("Error performing put request {}: {}", path, source))]
-    PutRequest {
+    #[snafu(display("Error performing request {}: {}", path, source))]
+    Request {
         source: crate::client::retry::Error,
         path: String,
     },
@@ -119,9 +114,9 @@ enum Error {
 impl From<Error> for crate::Error {
     fn from(err: Error) -> Self {
         match err {
-            Error::GetRequest { source, path }
-            | Error::DeleteRequest { source, path }
-            | Error::PutRequest { source, path } => source.error(STORE, path),
+            Error::GetRequest { source, path } | Error::Request { source, path } => {
+                source.error(STORE, path)
+            }
             _ => Self::Generic {
                 store: STORE,
                 source: Box::new(err),
@@ -170,15 +165,15 @@ impl GoogleCloudStorageConfig {
 }
 
 /// A builder for a put request allowing customisation of the headers and query string
-pub struct PutRequest<'a> {
+pub struct Request<'a> {
     path: &'a Path,
     config: &'a GoogleCloudStorageConfig,
-    payload: PutPayload,
+    payload: Option<PutPayload>,
     builder: RequestBuilder,
     idempotent: bool,
 }
 
-impl<'a> PutRequest<'a> {
+impl<'a> Request<'a> {
     fn header(self, k: &HeaderName, v: &str) -> Self {
         let builder = self.builder.header(k, v);
         Self { builder, ..self }
@@ -189,26 +184,58 @@ impl<'a> PutRequest<'a> {
         Self { builder, ..self }
     }
 
-    fn set_idempotent(mut self, idempotent: bool) -> Self {
+    fn idempotent(mut self, idempotent: bool) -> Self {
         self.idempotent = idempotent;
         self
     }
 
-    async fn send(self) -> Result<PutResult> {
+    fn with_attributes(self, attributes: Attributes) -> Self {
+        let mut builder = self.builder;
+        let mut has_content_type = false;
+        for (k, v) in &attributes {
+            builder = match k {
+                Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()),
+                Attribute::ContentType => {
+                    has_content_type = true;
+                    builder.header(CONTENT_TYPE, v.as_ref())
+                }
+            };
+        }
+
+        if !has_content_type {
+            let value = self.config.client_options.get_content_type(self.path);
+            builder = builder.header(CONTENT_TYPE, value.unwrap_or(DEFAULT_CONTENT_TYPE))
+        }
+        Self { builder, ..self }
+    }
+
+    fn with_payload(self, payload: PutPayload) -> Self {
+        let content_length = payload.content_length();
+        Self {
+            builder: self.builder.header(CONTENT_LENGTH, content_length),
+            payload: Some(payload),
+            ..self
+        }
+    }
+
+    async fn send(self) -> Result<Response> {
         let credential = self.config.credentials.get_credential().await?;
-        let response = self
+        let resp = self
             .builder
             .bearer_auth(&credential.bearer)
-            .header(CONTENT_LENGTH, self.payload.content_length())
             .retryable(&self.config.retry_config)
             .idempotent(self.idempotent)
-            .payload(Some(self.payload))
+            .payload(self.payload)
             .send()
             .await
-            .context(PutRequestSnafu {
+            .context(RequestSnafu {
                 path: self.path.as_ref(),
             })?;
+        Ok(resp)
+    }
 
+    async fn do_put(self) -> Result<PutResult> {
+        let response = self.send().await?;
         Ok(get_put_result(response.headers(), VERSION_HEADER).context(MetadataSnafu)?)
     }
 }
@@ -323,24 +350,13 @@ impl GoogleCloudStorageClient {
     /// Perform a put request <https://cloud.google.com/storage/docs/xml-api/put-object-upload>
     ///
     /// Returns the new ETag
-    pub fn put_request<'a>(&'a self, path: &'a Path, payload: PutPayload) -> PutRequest<'a> {
-        let url = self.object_url(path);
-
-        let content_type = self
-            .config
-            .client_options
-            .get_content_type(path)
-            .unwrap_or("application/octet-stream");
+    pub fn request<'a>(&'a self, method: Method, path: &'a Path) -> Request<'a> {
+        let builder = self.client.request(method, self.object_url(path));
 
-        let builder = self
-            .client
-            .request(Method::PUT, url)
-            .header(header::CONTENT_TYPE, content_type);
-
-        PutRequest {
+        Request {
             path,
             builder,
-            payload,
+            payload: None,
             config: &self.config,
             idempotent: false,
         }
@@ -352,10 +368,13 @@ impl GoogleCloudStorageClient {
         payload: PutPayload,
         opts: PutOptions,
     ) -> Result<PutResult> {
-        let builder = self.put_request(path, payload);
+        let builder = self
+            .request(Method::PUT, path)
+            .with_payload(payload)
+            .with_attributes(opts.attributes);
 
         let builder = match &opts.mode {
-            PutMode::Overwrite => builder.set_idempotent(true),
+            PutMode::Overwrite => builder.idempotent(true),
             PutMode::Create => builder.header(&VERSION_MATCH, "0"),
             PutMode::Update(v) => {
                 let etag = v.version.as_ref().context(MissingVersionSnafu)?;
@@ -363,7 +382,7 @@ impl GoogleCloudStorageClient {
             }
         };
 
-        match (opts.mode, builder.send().await) {
+        match (opts.mode, builder.do_put().await) {
             (PutMode::Create, Err(crate::Error::Precondition { path, source })) => {
                 Err(crate::Error::AlreadyExists { path, source })
             }
@@ -386,10 +405,11 @@ impl GoogleCloudStorageClient {
             ("uploadId", upload_id),
         ];
         let result = self
-            .put_request(path, data)
+            .request(Method::PUT, path)
+            .with_payload(data)
             .query(query)
-            .set_idempotent(true)
-            .send()
+            .idempotent(true)
+            .do_put()
             .await?;
 
         Ok(PartId {
@@ -398,30 +418,18 @@ impl GoogleCloudStorageClient {
     }
 
     /// Initiate a multipart upload <https://cloud.google.com/storage/docs/xml-api/post-object-multipart>
-    pub async fn multipart_initiate(&self, path: &Path) -> Result<MultipartId> {
-        let credential = self.get_credential().await?;
-        let url = self.object_url(path);
-
-        let content_type = self
-            .config
-            .client_options
-            .get_content_type(path)
-            .unwrap_or("application/octet-stream");
-
+    pub async fn multipart_initiate(
+        &self,
+        path: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<MultipartId> {
         let response = self
-            .client
-            .request(Method::POST, &url)
-            .bearer_auth(&credential.bearer)
-            .header(header::CONTENT_TYPE, content_type)
-            .header(header::CONTENT_LENGTH, "0")
+            .request(Method::POST, path)
+            .with_attributes(opts.attributes)
+            .header(&CONTENT_LENGTH, "0")
             .query(&[("uploads", "")])
-            .retryable(&self.config.retry_config)
-            .idempotent(true)
             .send()
-            .await
-            .context(PutRequestSnafu {
-                path: path.as_ref(),
-            })?;
+            .await?;
 
         let data = response.bytes().await.context(PutResponseBodySnafu)?;
         let result: InitiateMultipartUploadResult =
@@ -438,12 +446,12 @@ impl GoogleCloudStorageClient {
         self.client
             .request(Method::DELETE, &url)
             .bearer_auth(&credential.bearer)
-            .header(header::CONTENT_TYPE, "application/octet-stream")
-            .header(header::CONTENT_LENGTH, "0")
+            .header(CONTENT_TYPE, "application/octet-stream")
+            .header(CONTENT_LENGTH, "0")
             .query(&[("uploadId", multipart_id)])
             .send_retry(&self.config.retry_config)
             .await
-            .context(PutRequestSnafu {
+            .context(RequestSnafu {
                 path: path.as_ref(),
             })?;
 
@@ -459,9 +467,9 @@ impl GoogleCloudStorageClient {
         if completed_parts.is_empty() {
             // GCS doesn't allow empty multipart uploads
             let result = self
-                .put_request(path, Default::default())
-                .set_idempotent(true)
-                .send()
+                .request(Method::PUT, path)
+                .idempotent(true)
+                .do_put()
                 .await?;
             self.multipart_cleanup(path, multipart_id).await?;
             return Ok(result);
@@ -510,18 +518,7 @@ impl GoogleCloudStorageClient {
 
     /// Perform a delete request <https://cloud.google.com/storage/docs/xml-api/delete-object>
     pub async fn delete_request(&self, path: &Path) -> Result<()> {
-        let credential = self.get_credential().await?;
-        let url = self.object_url(path);
-
-        let builder = self.client.request(Method::DELETE, url);
-        builder
-            .bearer_auth(&credential.bearer)
-            .send_retry(&self.config.retry_config)
-            .await
-            .context(DeleteRequestSnafu {
-                path: path.as_ref(),
-            })?;
-
+        self.request(Method::DELETE, path).send().await?;
         Ok(())
     }
 
diff --git a/object_store/src/gcp/mod.rs b/object_store/src/gcp/mod.rs
index 149da76f5..0ec6e7e82 100644
--- a/object_store/src/gcp/mod.rs
+++ b/object_store/src/gcp/mod.rs
@@ -42,7 +42,8 @@ use crate::gcp::credential::GCSAuthorizer;
 use crate::signer::Signer;
 use crate::{
     multipart::PartId, path::Path, GetOptions, GetResult, ListResult, MultipartId, MultipartUpload,
-    ObjectMeta, ObjectStore, PutOptions, PutPayload, PutResult, Result, UploadPart,
+    ObjectMeta, ObjectStore, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
+    UploadPart,
 };
 use async_trait::async_trait;
 use client::GoogleCloudStorageClient;
@@ -156,8 +157,12 @@ impl ObjectStore for GoogleCloudStorage {
         self.client.put(location, payload, opts).await
     }
 
-    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
-        let upload_id = self.client.multipart_initiate(location).await?;
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        let upload_id = self.client.multipart_initiate(location, opts).await?;
 
         Ok(Box::new(GCSMultipartUpload {
             part_idx: 0,
@@ -206,7 +211,9 @@ impl ObjectStore for GoogleCloudStorage {
 #[async_trait]
 impl MultipartStore for GoogleCloudStorage {
     async fn create_multipart(&self, path: &Path) -> Result<MultipartId> {
-        self.client.multipart_initiate(path).await
+        self.client
+            .multipart_initiate(path, PutMultipartOpts::default())
+            .await
     }
 
     async fn put_part(
@@ -292,6 +299,8 @@ mod test {
             // Fake GCS server doesn't currently honor preconditions
             get_opts(&integration).await;
             put_opts(&integration, true).await;
+            // Fake GCS server doesn't currently support attributes
+            put_get_attributes(&integration).await;
         }
     }
 
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index 39f68ece6..cf259196b 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -21,11 +21,11 @@ use crate::client::retry::{self, RetryConfig, RetryExt};
 use crate::client::GetOptionsExt;
 use crate::path::{Path, DELIMITER};
 use crate::util::deserialize_rfc1123;
-use crate::{ClientOptions, GetOptions, ObjectMeta, PutPayload, Result};
+use crate::{Attribute, Attributes, ClientOptions, GetOptions, ObjectMeta, PutPayload, Result};
 use async_trait::async_trait;
 use bytes::Buf;
 use chrono::{DateTime, Utc};
-use hyper::header::CONTENT_LENGTH;
+use hyper::header::{CACHE_CONTROL, CONTENT_LENGTH};
 use percent_encoding::percent_decode_str;
 use reqwest::header::CONTENT_TYPE;
 use reqwest::{Method, Response, StatusCode};
@@ -157,13 +157,32 @@ impl Client {
         Ok(())
     }
 
-    pub async fn put(&self, location: &Path, payload: PutPayload) -> Result<Response> {
+    pub async fn put(
+        &self,
+        location: &Path,
+        payload: PutPayload,
+        attributes: Attributes,
+    ) -> Result<Response> {
         let mut retry = false;
         loop {
             let url = self.path_url(location);
             let mut builder = self.client.put(url);
-            if let Some(value) = self.client_options.get_content_type(location) {
-                builder = builder.header(CONTENT_TYPE, value);
+
+            let mut has_content_type = false;
+            for (k, v) in &attributes {
+                builder = match k {
+                    Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()),
+                    Attribute::ContentType => {
+                        has_content_type = true;
+                        builder.header(CONTENT_TYPE, v.as_ref())
+                    }
+                };
+            }
+
+            if !has_content_type {
+                if let Some(value) = self.client_options.get_content_type(location) {
+                    builder = builder.header(CONTENT_TYPE, value);
+                }
             }
 
             let resp = builder
diff --git a/object_store/src/http/mod.rs b/object_store/src/http/mod.rs
index a838a0f47..404211e57 100644
--- a/object_store/src/http/mod.rs
+++ b/object_store/src/http/mod.rs
@@ -44,7 +44,7 @@ use crate::http::client::Client;
 use crate::path::Path;
 use crate::{
     ClientConfigKey, ClientOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta,
-    ObjectStore, PutMode, PutOptions, PutPayload, PutResult, Result, RetryConfig,
+    ObjectStore, PutMode, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, RetryConfig,
 };
 
 mod client;
@@ -105,7 +105,7 @@ impl ObjectStore for HttpStore {
             return Err(crate::Error::NotImplemented);
         }
 
-        let response = self.client.put(location, payload).await?;
+        let response = self.client.put(location, payload, opts.attributes).await?;
         let e_tag = match get_etag(response.headers()) {
             Ok(e_tag) => Some(e_tag),
             Err(crate::client::header::Error::MissingEtag) => None,
@@ -118,7 +118,11 @@ impl ObjectStore for HttpStore {
         })
     }
 
-    async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
+    async fn put_multipart_opts(
+        &self,
+        _location: &Path,
+        _opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
         Err(crate::Error::NotImplemented)
     }
 
@@ -260,7 +264,7 @@ mod tests {
             .build()
             .unwrap();
 
-        put_get_delete_list_opts(&integration).await;
+        put_get_delete_list(&integration).await;
         list_uses_directories_correctly(&integration).await;
         list_with_delimiter(&integration).await;
         rename_and_copy(&integration).await;
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 157852ff9..ad72bd29e 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -543,6 +543,10 @@ mod payload;
 mod upload;
 mod util;
 
+mod attributes;
+
+pub use attributes::*;
+
 pub use parse::{parse_url, parse_url_opts};
 pub use payload::*;
 pub use upload::*;
@@ -593,7 +597,20 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
     ///
     /// Client should prefer [`ObjectStore::put`] for small payloads, as streaming uploads
     /// typically require multiple separate requests. See [`MultipartUpload`] for more information
-    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>>;
+    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
+        self.put_multipart_opts(location, PutMultipartOpts::default())
+            .await
+    }
+
+    /// Perform a multipart upload with options
+    ///
+    /// Client should prefer [`ObjectStore::put`] for small payloads, as streaming uploads
+    /// typically require multiple separate requests. See [`MultipartUpload`] for more information
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>>;
 
     /// Return the bytes that are stored at the specified location.
     async fn get(&self, location: &Path) -> Result<GetResult> {
@@ -781,6 +798,14 @@ macro_rules! as_ref_impl {
                 self.as_ref().put_multipart(location).await
             }
 
+            async fn put_multipart_opts(
+                &self,
+                location: &Path,
+                opts: PutMultipartOpts,
+            ) -> Result<Box<dyn MultipartUpload>> {
+                self.as_ref().put_multipart_opts(location, opts).await
+            }
+
             async fn get(&self, location: &Path) -> Result<GetResult> {
                 self.as_ref().get(location).await
             }
@@ -989,6 +1014,8 @@ pub struct GetResult {
     pub meta: ObjectMeta,
     /// The range of bytes returned by this request
     pub range: Range<usize>,
+    /// Additional object attributes
+    pub attributes: Attributes,
 }
 
 /// The kind of a [`GetResult`]
@@ -1114,6 +1141,10 @@ pub struct PutOptions {
     ///
     /// Implementations that don't support object tagging should ignore this
     pub tags: TagSet,
+    /// Provide a set of [`Attributes`]
+    ///
+    /// Implementations that don't support an attribute should return an error
+    pub attributes: Attributes,
 }
 
 impl From<PutMode> for PutOptions {
@@ -1134,6 +1165,46 @@ impl From<TagSet> for PutOptions {
     }
 }
 
+impl From<Attributes> for PutOptions {
+    fn from(attributes: Attributes) -> Self {
+        Self {
+            attributes,
+            ..Default::default()
+        }
+    }
+}
+
+/// Options for [`ObjectStore::put_multipart_opts`]
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct PutMultipartOpts {
+    /// Provide a [`TagSet`] for this object
+    ///
+    /// Implementations that don't support object tagging should ignore this
+    pub tags: TagSet,
+    /// Provide a set of [`Attributes`]
+    ///
+    /// Implementations that don't support an attribute should return an error
+    pub attributes: Attributes,
+}
+
+impl From<TagSet> for PutMultipartOpts {
+    fn from(tags: TagSet) -> Self {
+        Self {
+            tags,
+            ..Default::default()
+        }
+    }
+}
+
+impl From<Attributes> for PutMultipartOpts {
+    fn from(attributes: Attributes) -> Self {
+        Self {
+            attributes,
+            ..Default::default()
+        }
+    }
+}
+
 /// Result for a put request
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct PutResult {
@@ -1251,10 +1322,6 @@ mod tests {
     use rand::{thread_rng, Rng};
 
     pub(crate) async fn put_get_delete_list(storage: &DynObjectStore) {
-        put_get_delete_list_opts(storage).await
-    }
-
-    pub(crate) async fn put_get_delete_list_opts(storage: &DynObjectStore) {
         delete_fixtures(storage).await;
 
         let content_list = flatten_list_stream(storage, None).await.unwrap();
@@ -1674,6 +1741,38 @@ mod tests {
         storage.delete(&path).await.unwrap();
     }
 
+    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"),
+        ]);
+
+        let path = Path::from("attributes");
+        let opts = attributes.clone().into();
+        match integration.put_opts(&path, "foo".into(), opts).await {
+            Ok(_) => {
+                let r = integration.get(&path).await.unwrap();
+                assert_eq!(r.attributes, attributes);
+            }
+            Err(Error::NotImplemented) => {}
+            Err(e) => panic!("{e}"),
+        }
+
+        let opts = attributes.clone().into();
+        match integration.put_multipart_opts(&path, opts).await {
+            Ok(mut w) => {
+                w.put_part("foo".into()).await.unwrap();
+                w.complete().await.unwrap();
+
+                let r = integration.get(&path).await.unwrap();
+                assert_eq!(r.attributes, attributes);
+            }
+            Err(Error::NotImplemented) => {}
+            Err(e) => panic!("{e}"),
+        }
+    }
+
     pub(crate) async fn get_opts(storage: &dyn ObjectStore) {
         let path = Path::from("test");
         storage.put(&path, "foo".into()).await.unwrap();
@@ -2304,21 +2403,32 @@ mod tests {
 
         let path = Path::from("tag_test");
         storage
-            .put_opts(&path, "test".into(), tag_set.into())
+            .put_opts(&path, "test".into(), tag_set.clone().into())
             .await
             .unwrap();
 
+        let multi_path = Path::from("tag_test_multi");
+        let mut write = storage
+            .put_multipart_opts(&multi_path, tag_set.into())
+            .await
+            .unwrap();
+
+        write.put_part("foo".into()).await.unwrap();
+        write.complete().await.unwrap();
+
         // Write should always succeed, but certain configurations may simply ignore tags
         if !validate {
             return;
         }
 
-        let resp = get_tags(path.clone()).await.unwrap();
-        let body = resp.bytes().await.unwrap();
+        for path in [path, multi_path] {
+            let resp = get_tags(path.clone()).await.unwrap();
+            let body = resp.bytes().await.unwrap();
 
-        let mut resp: Tagging = quick_xml::de::from_reader(body.reader()).unwrap();
-        resp.list.tags.sort_by(|a, b| a.key.cmp(&b.key));
-        assert_eq!(resp.list.tags, tags);
+            let mut resp: Tagging = quick_xml::de::from_reader(body.reader()).unwrap();
+            resp.list.tags.sort_by(|a, b| a.key.cmp(&b.key));
+            assert_eq!(resp.list.tags, tags);
+        }
     }
 
     async fn delete_fixtures(storage: &DynObjectStore) {
diff --git a/object_store/src/limit.rs b/object_store/src/limit.rs
index b94aa05b8..f3e1d4296 100644
--- a/object_store/src/limit.rs
+++ b/object_store/src/limit.rs
@@ -19,7 +19,8 @@
 
 use crate::{
     BoxStream, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
-    ObjectStore, Path, PutOptions, PutPayload, PutResult, Result, StreamExt, UploadPart,
+    ObjectStore, Path, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, StreamExt,
+    UploadPart,
 };
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -91,6 +92,19 @@ impl<T: ObjectStore> ObjectStore for LimitStore<T> {
             upload,
         }))
     }
+
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        let upload = self.inner.put_multipart_opts(location, opts).await?;
+        Ok(Box::new(LimitUpload {
+            semaphore: Arc::clone(&self.semaphore),
+            upload,
+        }))
+    }
+
     async fn get(&self, location: &Path) -> Result<GetResult> {
         let permit = Arc::clone(&self.semaphore).acquire_owned().await.unwrap();
         let r = self.inner.get(location).await?;
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index d5581cdc8..8dec5bee0 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -38,8 +38,8 @@ use crate::{
     maybe_spawn_blocking,
     path::{absolute_path_to_url, Path},
     util::InvalidGetRange,
-    GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
-    PutMode, PutOptions, PutPayload, PutResult, Result, UploadPart,
+    Attributes, GetOptions, GetResult, GetResultPayload, ListResult, MultipartUpload, ObjectMeta,
+    ObjectStore, PutMode, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, UploadPart,
 };
 
 /// A specialized `Error` for filesystem object store-related errors
@@ -346,6 +346,10 @@ impl ObjectStore for LocalFileSystem {
             return Err(crate::Error::NotImplemented);
         }
 
+        if !opts.attributes.is_empty() {
+            return Err(crate::Error::NotImplemented);
+        }
+
         let path = self.path_to_filesystem(location)?;
         maybe_spawn_blocking(move || {
             let (mut file, staging_path) = new_staged_upload(&path)?;
@@ -400,7 +404,15 @@ impl ObjectStore for LocalFileSystem {
         .await
     }
 
-    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        if !opts.attributes.is_empty() {
+            return Err(crate::Error::NotImplemented);
+        }
+
         let dest = self.path_to_filesystem(location)?;
         let (file, src) = new_staged_upload(&dest)?;
         Ok(Box::new(LocalUpload::new(src, dest, file)))
@@ -421,6 +433,7 @@ impl ObjectStore for LocalFileSystem {
 
             Ok(GetResult {
                 payload: GetResultPayload::File(file, path),
+                attributes: Attributes::default(),
                 range,
                 meta,
             })
diff --git a/object_store/src/memory.rs b/object_store/src/memory.rs
index d42e6f231..daf14e175 100644
--- a/object_store/src/memory.rs
+++ b/object_store/src/memory.rs
@@ -30,8 +30,9 @@ use snafu::{OptionExt, ResultExt, Snafu};
 use crate::multipart::{MultipartStore, PartId};
 use crate::util::InvalidGetRange;
 use crate::{
-    path::Path, GetRange, GetResult, GetResultPayload, ListResult, MultipartId, MultipartUpload,
-    ObjectMeta, ObjectStore, PutMode, PutOptions, PutResult, Result, UpdateVersion, UploadPart,
+    path::Path, Attributes, GetRange, GetResult, GetResultPayload, ListResult, MultipartId,
+    MultipartUpload, ObjectMeta, ObjectStore, PutMode, PutMultipartOpts, PutOptions, PutResult,
+    Result, UpdateVersion, UploadPart,
 };
 use crate::{GetOptions, PutPayload};
 
@@ -88,15 +89,22 @@ pub struct InMemory {
 struct Entry {
     data: Bytes,
     last_modified: DateTime<Utc>,
+    attributes: Attributes,
     e_tag: usize,
 }
 
 impl Entry {
-    fn new(data: Bytes, last_modified: DateTime<Utc>, e_tag: usize) -> Self {
+    fn new(
+        data: Bytes,
+        last_modified: DateTime<Utc>,
+        e_tag: usize,
+        attributes: Attributes,
+    ) -> Self {
         Self {
             data,
             last_modified,
             e_tag,
+            attributes,
         }
     }
 }
@@ -116,10 +124,10 @@ struct PartStorage {
 type SharedStorage = Arc<RwLock<Storage>>;
 
 impl Storage {
-    fn insert(&mut self, location: &Path, bytes: Bytes) -> usize {
+    fn insert(&mut self, location: &Path, bytes: Bytes, attributes: Attributes) -> usize {
         let etag = self.next_etag;
         self.next_etag += 1;
-        let entry = Entry::new(bytes, Utc::now(), etag);
+        let entry = Entry::new(bytes, Utc::now(), etag, attributes);
         self.overwrite(location, entry);
         etag
     }
@@ -200,7 +208,7 @@ impl ObjectStore for InMemory {
     ) -> Result<PutResult> {
         let mut storage = self.storage.write();
         let etag = storage.next_etag;
-        let entry = Entry::new(payload.into(), Utc::now(), etag);
+        let entry = Entry::new(payload.into(), Utc::now(), etag, opts.attributes);
 
         match opts.mode {
             PutMode::Overwrite => storage.overwrite(location, entry),
@@ -215,9 +223,14 @@ impl ObjectStore for InMemory {
         })
     }
 
-    async fn put_multipart(&self, location: &Path) -> Result<Box<dyn MultipartUpload>> {
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
         Ok(Box::new(InMemoryUpload {
             location: location.clone(),
+            attributes: opts.attributes,
             parts: vec![],
             storage: Arc::clone(&self.storage),
         }))
@@ -247,6 +260,7 @@ impl ObjectStore for InMemory {
 
         Ok(GetResult {
             payload: GetResultPayload::Stream(stream.boxed()),
+            attributes: entry.attributes,
             meta,
             range,
         })
@@ -363,7 +377,9 @@ impl ObjectStore for InMemory {
 
     async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
         let entry = self.entry(from).await?;
-        self.storage.write().insert(to, entry.data);
+        self.storage
+            .write()
+            .insert(to, entry.data, entry.attributes);
         Ok(())
     }
 
@@ -376,7 +392,7 @@ impl ObjectStore for InMemory {
             }
             .into());
         }
-        storage.insert(to, entry.data);
+        storage.insert(to, entry.data, entry.attributes);
         Ok(())
     }
 }
@@ -426,7 +442,7 @@ impl MultipartStore for InMemory {
         for x in &upload.parts {
             buf.extend_from_slice(x.as_ref().unwrap())
         }
-        let etag = storage.insert(path, buf.into());
+        let etag = storage.insert(path, buf.into(), Default::default());
         Ok(PutResult {
             e_tag: Some(etag.to_string()),
             version: None,
@@ -476,6 +492,7 @@ impl InMemory {
 #[derive(Debug)]
 struct InMemoryUpload {
     location: Path,
+    attributes: Attributes,
     parts: Vec<PutPayload>,
     storage: Arc<RwLock<Storage>>,
 }
@@ -492,7 +509,12 @@ impl MultipartUpload for InMemoryUpload {
         let mut buf = Vec::with_capacity(cap);
         let parts = self.parts.iter().flatten();
         parts.for_each(|x| buf.extend_from_slice(x));
-        let etag = self.storage.write().insert(&self.location, buf.into());
+        let etag = self.storage.write().insert(
+            &self.location,
+            buf.into(),
+            std::mem::take(&mut self.attributes),
+        );
+
         Ok(PutResult {
             e_tag: Some(etag.to_string()),
             version: None,
@@ -523,6 +545,7 @@ mod tests {
         stream_get(&integration).await;
         put_opts(&integration, true).await;
         multipart(&integration, &integration).await;
+        put_get_attributes(&integration).await;
     }
 
     #[tokio::test]
diff --git a/object_store/src/prefix.rs b/object_store/src/prefix.rs
index 1d1ffeed8..7c9ea5804 100644
--- a/object_store/src/prefix.rs
+++ b/object_store/src/prefix.rs
@@ -22,8 +22,8 @@ use std::ops::Range;
 
 use crate::path::Path;
 use crate::{
-    GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, PutOptions,
-    PutPayload, PutResult, Result,
+    GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, PutMultipartOpts,
+    PutOptions, PutPayload, PutResult, Result,
 };
 
 #[doc(hidden)]
@@ -100,6 +100,15 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
         self.inner.put_multipart(&full_path).await
     }
 
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        let full_path = self.full_path(location);
+        self.inner.put_multipart_opts(&full_path, opts).await
+    }
+
     async fn get(&self, location: &Path) -> Result<GetResult> {
         let full_path = self.full_path(location);
         self.inner.get(&full_path).await
diff --git a/object_store/src/throttle.rs b/object_store/src/throttle.rs
index d08978466..38b6d7c3b 100644
--- a/object_store/src/throttle.rs
+++ b/object_store/src/throttle.rs
@@ -23,7 +23,7 @@ use std::{convert::TryInto, sync::Arc};
 use crate::multipart::{MultipartStore, PartId};
 use crate::{
     path::Path, GetResult, GetResultPayload, ListResult, MultipartId, MultipartUpload, ObjectMeta,
-    ObjectStore, PutOptions, PutPayload, PutResult, Result,
+    ObjectStore, PutMultipartOpts, PutOptions, PutPayload, PutResult, Result,
 };
 use crate::{GetOptions, UploadPart};
 use async_trait::async_trait;
@@ -171,6 +171,18 @@ impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
         }))
     }
 
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
+        let upload = self.inner.put_multipart_opts(location, opts).await?;
+        Ok(Box::new(ThrottledUpload {
+            upload,
+            sleep: self.config().wait_put_per_call,
+        }))
+    }
+
     async fn get(&self, location: &Path) -> Result<GetResult> {
         sleep(self.config().wait_get_per_call).await;
 
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/object_store/tests/get_range_file.rs b/object_store/tests/get_range_file.rs
index 59c593400..c5550ac21 100644
--- a/object_store/tests/get_range_file.rs
+++ b/object_store/tests/get_range_file.rs
@@ -46,7 +46,11 @@ impl ObjectStore for MyStore {
         self.0.put_opts(location, payload, opts).await
     }
 
-    async fn put_multipart(&self, _location: &Path) -> Result<Box<dyn MultipartUpload>> {
+    async fn put_multipart_opts(
+        &self,
+        _location: &Path,
+        _opts: PutMultipartOpts,
+    ) -> Result<Box<dyn MultipartUpload>> {
         todo!()
     }