Repository: apache/arrow-rs
Pull request #5704 (2024-05-01 07:07:35)

Support casting `StringView`/`BinaryView` --> `StringArray`/`BinaryArray`.
# 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.
-->

Part of #5508.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

# What changes are included in this PR?

<!--
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.
-->

Support casting from byte view array (BinaryViewArray, StringVIewArray) to byte array (BinaryArray, StringArray).

# Are there any user-facing changes?

No.

<!--
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.
-->


RinChanNOWWW: I have written some of my thoughts in #5508.

Patch:
diff --git a/arrow-array/src/array/null_array.rs b/arrow-array/src/array/null_array.rs
index af3ec0b57..88cc2d911 100644
--- a/arrow-array/src/array/null_array.rs
+++ b/arrow-array/src/array/null_array.rs
@@ -67,8 +67,11 @@ impl NullArray {
     }
 
     /// Returns a new null array builder
-    pub fn builder(capacity: usize) -> NullBuilder {
-        NullBuilder::with_capacity(capacity)
+    ///
+    /// Note that the `capacity` parameter to this function is _deprecated_. It
+    /// now does nothing, and will be removed in a future version.
+    pub fn builder(_capacity: usize) -> NullBuilder {
+        NullBuilder::new()
     }
 }
 
diff --git a/arrow-array/src/builder/generic_list_builder.rs b/arrow-array/src/builder/generic_list_builder.rs
index 25903bcf5..d6fff12e1 100644
--- a/arrow-array/src/builder/generic_list_builder.rs
+++ b/arrow-array/src/builder/generic_list_builder.rs
@@ -548,7 +548,7 @@ mod tests {
     }
 
     #[test]
-    fn test_boxed_primitive_aray_builder() {
+    fn test_boxed_primitive_array_builder() {
         let values_builder = make_builder(&DataType::Int32, 5);
         let mut builder = ListBuilder::new(values_builder);
 
diff --git a/arrow-array/src/builder/null_builder.rs b/arrow-array/src/builder/null_builder.rs
index 53a6b103d..59086dffa 100644
--- a/arrow-array/src/builder/null_builder.rs
+++ b/arrow-array/src/builder/null_builder.rs
@@ -60,11 +60,13 @@ impl NullBuilder {
     }
 
     /// Creates a new null builder with space for `capacity` elements without re-allocating
-    pub fn with_capacity(capacity: usize) -> Self {
-        Self { len: capacity }
+    #[deprecated = "there is no actual notion of capacity in the NullBuilder, so emulating it makes little sense"]
+    pub fn with_capacity(_capacity: usize) -> Self {
+        Self::new()
     }
 
     /// Returns the capacity of this builder measured in slots of type `T`
+    #[deprecated = "there is no actual notion of capacity in the NullBuilder, so emulating it makes little sense"]
     pub fn capacity(&self) -> usize {
         self.len
     }
@@ -158,7 +160,7 @@ mod tests {
         builder.append_empty_values(4);
 
         let arr = builder.finish();
-        assert_eq!(20, arr.len());
+        assert_eq!(10, arr.len());
         assert_eq!(0, arr.offset());
         assert_eq!(0, arr.null_count());
         assert!(arr.is_nullable());
@@ -171,10 +173,10 @@ mod tests {
         builder.append_empty_value();
         builder.append_empty_values(3);
         let mut array = builder.finish_cloned();
-        assert_eq!(21, array.len());
+        assert_eq!(5, array.len());
 
         builder.append_empty_values(5);
         array = builder.finish();
-        assert_eq!(26, array.len());
+        assert_eq!(10, array.len());
     }
 }
diff --git a/arrow-array/src/builder/struct_builder.rs b/arrow-array/src/builder/struct_builder.rs
index 1e2e402f7..46202bfa3 100644
--- a/arrow-array/src/builder/struct_builder.rs
+++ b/arrow-array/src/builder/struct_builder.rs
@@ -168,7 +168,7 @@ impl ArrayBuilder for StructBuilder {
 pub fn make_builder(datatype: &DataType, capacity: usize) -> Box<dyn ArrayBuilder> {
     use crate::builder::*;
     match datatype {
-        DataType::Null => Box::new(NullBuilder::with_capacity(capacity)),
+        DataType::Null => Box::new(NullBuilder::new()),
         DataType::Boolean => Box::new(BooleanBuilder::with_capacity(capacity)),
         DataType::Int8 => Box::new(Int8Builder::with_capacity(capacity)),
         DataType::Int16 => Box::new(Int16Builder::with_capacity(capacity)),
diff --git a/arrow-cast/src/cast/mod.rs b/arrow-cast/src/cast/mod.rs
index 36072760e..171267f80 100644
--- a/arrow-cast/src/cast/mod.rs
+++ b/arrow-cast/src/cast/mod.rs
@@ -218,6 +218,8 @@ pub fn can_cast_types(from_type: &DataType, to_type: &DataType) -> bool {
             | Interval(_),
         ) => true,
         (Utf8 | LargeUtf8, Utf8View) => true,
+        (Utf8View, Utf8 | LargeUtf8) => true,
+        (BinaryView, Binary | LargeBinary) => true,
         (Utf8 | LargeUtf8, _) => to_type.is_numeric() && to_type != &Float16,
         (_, Utf8 | LargeUtf8) => from_type.is_primitive(),
 
@@ -1262,6 +1264,12 @@ pub fn cast_with_options(
                 "Casting from {from_type:?} to {to_type:?} not supported",
             ))),
         },
+        (Utf8View, Utf8) => cast_view_to_byte::<StringViewType, GenericStringType<i32>>(array),
+        (Utf8View, LargeUtf8) => cast_view_to_byte::<StringViewType, GenericStringType<i64>>(array),
+        (BinaryView, Binary) => cast_view_to_byte::<BinaryViewType, GenericBinaryType<i32>>(array),
+        (BinaryView, LargeBinary) => {
+            cast_view_to_byte::<BinaryViewType, GenericBinaryType<i64>>(array)
+        }
         (from_type, LargeUtf8) if from_type.is_primitive() => {
             value_to_string::<i64>(array, cast_options)
         }
@@ -2299,6 +2307,32 @@ where
     }))
 }
 
+/// Helper function to cast from one `ByteViewType` array to `ByteArrayType` array.
+fn cast_view_to_byte<FROM, TO>(array: &dyn Array) -> Result<ArrayRef, ArrowError>
+where
+    FROM: ByteViewType,
+    TO: ByteArrayType,
+    FROM::Native: AsRef<TO::Native>,
+{
+    let data = array.to_data();
+    let view_array = GenericByteViewArray::<FROM>::from(data);
+
+    let len = view_array.len();
+    let bytes = view_array
+        .views()
+        .iter()
+        .map(|v| ByteView::from(*v).length as usize)
+        .sum::<usize>();
+
+    let mut byte_array_builder = GenericByteBuilder::<TO>::with_capacity(len, bytes);
+
+    for val in view_array.iter() {
+        byte_array_builder.append_option(val);
+    }
+
+    Ok(Arc::new(byte_array_builder.finish()))
+}
+
 #[cfg(test)]
 mod tests {
     use arrow_buffer::{Buffer, NullBuffer};
@@ -5169,6 +5203,82 @@ mod tests {
         assert_eq!(binary_view_array.as_ref(), &expect_binary_view_array);
     }
 
+    #[test]
+    fn test_view_to_string() {
+        _test_view_to_string::<i32>();
+        _test_view_to_string::<i64>();
+    }
+
+    fn _test_view_to_string<O>()
+    where
+        O: OffsetSizeTrait,
+    {
+        let data: Vec<Option<&str>> = vec![
+            Some("hello"),
+            Some("world"),
+            None,
+            Some("large payload over 12 bytes"),
+            Some("lulu"),
+        ];
+
+        let view_array = {
+            // ["hello", "world", null, "large payload over 12 bytes", "lulu"]
+            let mut builder = StringViewBuilder::new().with_block_size(8); // multiple buffers.
+            for s in data.iter() {
+                builder.append_option(*s);
+            }
+            builder.finish()
+        };
+
+        let expected_string_array = GenericStringArray::<O>::from(data);
+        let expected_type = expected_string_array.data_type();
+
+        assert!(can_cast_types(view_array.data_type(), expected_type));
+
+        let string_array = cast(&view_array, expected_type).unwrap();
+        assert_eq!(string_array.data_type(), expected_type);
+
+        assert_eq!(string_array.as_ref(), &expected_string_array);
+    }
+
+    #[test]
+    fn test_view_to_binary() {
+        _test_view_to_binary::<i32>();
+        _test_view_to_binary::<i64>();
+    }
+
+    fn _test_view_to_binary<O>()
+    where
+        O: OffsetSizeTrait,
+    {
+        let data: Vec<Option<&[u8]>> = vec![
+            Some(b"hello"),
+            Some(b"world"),
+            None,
+            Some(b"large payload over 12 bytes"),
+            Some(b"lulu"),
+        ];
+
+        let view_array = {
+            // ["hello", "world", null, "large payload over 12 bytes", "lulu"]
+            let mut builder = BinaryViewBuilder::new().with_block_size(8); // multiple buffers.
+            for s in data.iter() {
+                builder.append_option(*s);
+            }
+            builder.finish()
+        };
+
+        let expected_binary_array = GenericBinaryArray::<O>::from(data);
+        let expected_type = expected_binary_array.data_type();
+
+        assert!(can_cast_types(view_array.data_type(), expected_type));
+
+        let binary_array = cast(&view_array, expected_type).unwrap();
+        assert_eq!(binary_array.data_type(), expected_type);
+
+        assert_eq!(binary_array.as_ref(), &expected_binary_array);
+    }
+
     #[test]
     fn test_cast_from_f64() {
         let f64_values: Vec<f64> = vec![
diff --git a/arrow-cast/src/parse.rs b/arrow-cast/src/parse.rs
index 3102fcc58..f48a309fd 100644
--- a/arrow-cast/src/parse.rs
+++ b/arrow-cast/src/parse.rs
@@ -435,11 +435,20 @@ impl Parser for Float64Type {
     }
 }
 
+/// This API is only stable since 1.70 so can't use it when current MSRV is lower
+#[inline(always)]
+fn is_some_and<T>(opt: Option<T>, f: impl FnOnce(T) -> bool) -> bool {
+    match opt {
+        None => false,
+        Some(x) => f(x),
+    }
+}
+
 macro_rules! parser_primitive {
     ($t:ty) => {
         impl Parser for $t {
             fn parse(string: &str) -> Option<Self::Native> {
-                if !string.as_bytes().last().is_some_and(|x| x.is_ascii_digit()) {
+                if !is_some_and(string.as_bytes().last(), |x| x.is_ascii_digit()) {
                     return None;
                 }
                 match atoi::FromRadix10SignedChecked::from_radix_10_signed_checked(
diff --git a/arrow-csv/src/reader/mod.rs b/arrow-csv/src/reader/mod.rs
index 943bd8905..4d7d60e10 100644
--- a/arrow-csv/src/reader/mod.rs
+++ b/arrow-csv/src/reader/mod.rs
@@ -125,7 +125,7 @@
 
 mod records;
 
-use arrow_array::builder::PrimitiveBuilder;
+use arrow_array::builder::{NullBuilder, PrimitiveBuilder};
 use arrow_array::types::*;
 use arrow_array::*;
 use arrow_cast::parse::{parse_decimal, string_to_datetime, Parser};
@@ -759,7 +759,11 @@ fn parse(
                         null_regex,
                     )
                 }
-                DataType::Null => Ok(Arc::new(NullArray::builder(rows.len()).finish()) as ArrayRef),
+                DataType::Null => Ok(Arc::new({
+                    let mut builder = NullBuilder::new();
+                    builder.append_nulls(rows.len());
+                    builder.finish()
+                }) as ArrayRef),
                 DataType::Utf8 => Ok(Arc::new(
                     rows.iter()
                         .map(|row| {
diff --git a/arrow-flight/src/client.rs b/arrow-flight/src/client.rs
index b2abfb0c1..a7e15fe24 100644
--- a/arrow-flight/src/client.rs
+++ b/arrow-flight/src/client.rs
@@ -417,9 +417,7 @@ impl FlightClient {
     ///
     /// // encode the batch as a stream of `FlightData`
     /// let flight_data_stream = FlightDataEncoderBuilder::new()
-    ///   .build(futures::stream::iter(vec![Ok(batch)]))
-    ///   // data encoder return Results, but do_exchange requires FlightData
-    ///   .map(|batch|batch.unwrap());
+    ///   .build(futures::stream::iter(vec![Ok(batch)]));
     ///
     /// // send the stream and get the results as `RecordBatches`
     /// let response: Vec<RecordBatch> = client
@@ -431,20 +429,40 @@ impl FlightClient {
     ///   .expect("error calling do_exchange");
     /// # }
     /// ```
-    pub async fn do_exchange<S: Stream<Item = FlightData> + Send + 'static>(
+    pub async fn do_exchange<S: Stream<Item = Result<FlightData>> + Send + 'static>(
         &mut self,
         request: S,
     ) -> Result<FlightRecordBatchStream> {
-        let request = self.make_request(request);
+        let (sender, mut receiver) = futures::channel::oneshot::channel();
 
-        let response = self
-            .inner
-            .do_exchange(request)
-            .await?
-            .into_inner()
-            .map_err(FlightError::Tonic);
+        // Intercepts client errors and sends them to the oneshot channel above
+        let mut request = Box::pin(request); // Pin to heap
+        let mut sender = Some(sender); // Wrap into Option so can be taken
+        let request_stream = futures::stream::poll_fn(move |cx| {
+            Poll::Ready(match ready!(request.poll_next_unpin(cx)) {
+                Some(Ok(data)) => Some(data),
+                Some(Err(e)) => {
+                    let _ = sender.take().unwrap().send(e);
+                    None
+                }
+                None => None,
+            })
+        });
+
+        let request = self.make_request(request_stream);
+        let mut response_stream = self.inner.do_exchange(request).await?.into_inner();
+
+        // Forwards errors from the error oneshot with priority over responses from server
+        let error_stream = futures::stream::poll_fn(move |cx| {
+            if let Poll::Ready(Ok(err)) = receiver.poll_unpin(cx) {
+                return Poll::Ready(Some(Err(err)));
+            }
+            let next = ready!(response_stream.poll_next_unpin(cx));
+            Poll::Ready(next.map(|x| x.map_err(FlightError::Tonic)))
+        });
 
-        Ok(FlightRecordBatchStream::new_from_flight_data(response))
+        // combine the response from the server and any error from the client
+        Ok(FlightRecordBatchStream::new_from_flight_data(error_stream))
     }
 
     /// Make a `ListFlights` call to the server with the provided
diff --git a/arrow-json/src/reader/serializer.rs b/arrow-json/src/reader/serializer.rs
index 378d77bd9..2e1d76f98 100644
--- a/arrow-json/src/reader/serializer.rs
+++ b/arrow-json/src/reader/serializer.rs
@@ -309,16 +309,16 @@ impl<'a, 'b> SerializeMap for ObjectSerializer<'a, 'b> {
     type Ok = ();
     type Error = SerializerError;
 
-    fn serialize_key<T: ?Sized>(&mut self, key: &T) -> Result<(), Self::Error>
+    fn serialize_key<T>(&mut self, key: &T) -> Result<(), Self::Error>
     where
-        T: Serialize,
+        T: Serialize + ?Sized,
     {
         key.serialize(&mut *self.serializer)
     }
 
-    fn serialize_value<T: ?Sized>(&mut self, value: &T) -> Result<(), Self::Error>
+    fn serialize_value<T>(&mut self, value: &T) -> Result<(), Self::Error>
     where
-        T: Serialize,
+        T: Serialize + ?Sized,
     {
         value.serialize(&mut *self.serializer)
     }
@@ -333,13 +333,9 @@ impl<'a, 'b> SerializeStruct for ObjectSerializer<'a, 'b> {
     type Ok = ();
     type Error = SerializerError;
 
-    fn serialize_field<T: ?Sized>(
-        &mut self,
-        key: &'static str,
-        value: &T,
-    ) -> Result<(), Self::Error>
+    fn serialize_field<T>(&mut self, key: &'static str, value: &T) -> Result<(), Self::Error>
     where
-        T: Serialize,
+        T: Serialize + ?Sized,
     {
         key.serialize(&mut *self.serializer)?;
         value.serialize(&mut *self.serializer)
@@ -376,9 +372,9 @@ impl<'a, 'b> SerializeSeq for ListSerializer<'a, 'b> {
     type Ok = ();
     type Error = SerializerError;
 
-    fn serialize_element<T: ?Sized>(&mut self, value: &T) -> Result<(), Self::Error>
+    fn serialize_element<T>(&mut self, value: &T) -> Result<(), Self::Error>
     where
-        T: Serialize,
+        T: Serialize + ?Sized,
     {
         value.serialize(&mut *self.serializer)
     }
@@ -393,9 +389,9 @@ impl<'a, 'b> SerializeTuple for ListSerializer<'a, 'b> {
     type Ok = ();
     type Error = SerializerError;
 
-    fn serialize_element<T: ?Sized>(&mut self, value: &T) -> Result<(), Self::Error>
+    fn serialize_element<T>(&mut self, value: &T) -> Result<(), Self::Error>
     where
-        T: Serialize,
+        T: Serialize + ?Sized,
     {
         value.serialize(&mut *self.serializer)
     }
@@ -410,9 +406,9 @@ impl<'a, 'b> SerializeTupleStruct for ListSerializer<'a, 'b> {
     type Ok = ();
     type Error = SerializerError;
 
-    fn serialize_field<T: ?Sized>(&mut self, value: &T) -> Result<(), Self::Error>
+    fn serialize_field<T>(&mut self, value: &T) -> Result<(), Self::Error>
     where
-        T: Serialize,
+        T: Serialize + ?Sized,
     {
         value.serialize(&mut *self.serializer)
     }
diff --git a/arrow-json/src/writer/encoder.rs b/arrow-json/src/writer/encoder.rs
index 810e65b22..056ddf3dd 100644
--- a/arrow-json/src/writer/encoder.rs
+++ b/arrow-json/src/writer/encoder.rs
@@ -155,12 +155,21 @@ struct StructArrayEncoder<'a> {
     explicit_nulls: bool,
 }
 
+/// This API is only stable since 1.70 so can't use it when current MSRV is lower
+#[inline(always)]
+fn is_some_and<T>(opt: Option<T>, f: impl FnOnce(T) -> bool) -> bool {
+    match opt {
+        None => false,
+        Some(x) => f(x),
+    }
+}
+
 impl<'a> Encoder for StructArrayEncoder<'a> {
     fn encode(&mut self, idx: usize, out: &mut Vec<u8>) {
         out.push(b'{');
         let mut is_first = true;
         for field_encoder in &mut self.encoders {
-            let is_null = field_encoder.nulls.as_ref().is_some_and(|n| n.is_null(idx));
+            let is_null = is_some_and(field_encoder.nulls.as_ref(), |n| n.is_null(idx));
             if is_null && !self.explicit_nulls {
                 continue;
             }
@@ -447,13 +456,13 @@ impl<'a> MapEncoder<'a> {
         let (values, value_nulls) = make_encoder_impl(values, options)?;
 
         // We sanity check nulls as these are currently not enforced by MapArray (#1697)
-        if key_nulls.is_some_and(|x| x.null_count() != 0) {
+        if is_some_and(key_nulls, |x| x.null_count() != 0) {
             return Err(ArrowError::InvalidArgumentError(
                 "Encountered nulls in MapArray keys".to_string(),
             ));
         }
 
-        if array.entries().nulls().is_some_and(|x| x.null_count() != 0) {
+        if is_some_and(array.entries().nulls(), |x| x.null_count() != 0) {
             return Err(ArrowError::InvalidArgumentError(
                 "Encountered nulls in MapArray entries".to_string(),
             ));
@@ -478,7 +487,7 @@ impl<'a> Encoder for MapEncoder<'a> {
 
         out.push(b'{');
         for idx in start..end {
-            let is_null = self.value_nulls.as_ref().is_some_and(|n| n.is_null(idx));
+            let is_null = is_some_and(self.value_nulls.as_ref(), |n| n.is_null(idx));
             if is_null && !self.explicit_nulls {
                 continue;
             }
diff --git a/arrow-select/src/take.rs b/arrow-select/src/take.rs
index dc9e13040..8939d3f71 100644
--- a/arrow-select/src/take.rs
+++ b/arrow-select/src/take.rs
@@ -1401,9 +1401,9 @@ mod tests {
         );
     }
 
-    fn _test_take_string<'a, K: 'static>()
+    fn _test_take_string<'a, K>()
     where
-        K: Array + PartialEq + From<Vec<Option<&'a str>>>,
+        K: Array + PartialEq + From<Vec<Option<&'a str>>> + 'static,
     {
         let index = UInt32Array::from(vec![Some(3), None, Some(1), Some(3), Some(4)]);
 
diff --git a/arrow/src/util/string_writer.rs b/arrow/src/util/string_writer.rs
index 4c61f183e..57a98d6f5 100644
--- a/arrow/src/util/string_writer.rs
+++ b/arrow/src/util/string_writer.rs
@@ -63,6 +63,7 @@
 //! }
 //! ```
 
+use std::fmt::Formatter;
 use std::io::{Error, ErrorKind, Result, Write};
 
 #[derive(Debug)]
@@ -83,10 +84,9 @@ impl Default for StringWriter {
         Self::new()
     }
 }
-
-impl ToString for StringWriter {
-    fn to_string(&self) -> String {
-        self.data.clone()
+impl std::fmt::Display for StringWriter {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.data)
     }
 }
 
diff --git a/object_store/Cargo.toml b/object_store/Cargo.toml
index ca1e5b327..c61946f7f 100644
--- a/object_store/Cargo.toml
+++ b/object_store/Cargo.toml
@@ -66,6 +66,7 @@ gcp = ["cloud", "rustls-pemfile"]
 aws = ["cloud", "md-5"]
 http = ["cloud"]
 tls-webpki-roots = ["reqwest?/rustls-tls-webpki-roots"]
+integration = []
 
 [dev-dependencies] # In alphabetical order
 futures-test = "0.3"
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index 664e18364..1e42093fd 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -70,21 +70,9 @@ enum Error {
     #[snafu(display("Configuration key: '{}' is not known.", key))]
     UnknownConfigurationKey { key: String },
 
-    #[snafu(display("Bucket '{}' not found", bucket))]
-    BucketNotFound { bucket: String },
-
-    #[snafu(display("Failed to resolve region for bucket '{}'", bucket))]
-    ResolveRegion {
-        bucket: String,
-        source: reqwest::Error,
-    },
-
     #[snafu(display("Invalid Zone suffix for bucket '{bucket}'"))]
     ZoneSuffix { bucket: String },
 
-    #[snafu(display("Failed to parse the region for bucket '{}'", bucket))]
-    RegionParse { bucket: String },
-
     #[snafu(display("Invalid encryption type: {}. Valid values are \"AES256\", \"sse:kms\", and \"sse:kms:dsse\".", passed))]
     InvalidEncryptionType { passed: String },
 
diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 24247688e..98226c4b6 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -66,12 +66,6 @@ const SHA256_CHECKSUM: &str = "x-amz-checksum-sha256";
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
 pub(crate) enum Error {
-    #[snafu(display("Error fetching get response body {}: {}", path, source))]
-    GetResponseBody {
-        source: reqwest::Error,
-        path: String,
-    },
-
     #[snafu(display("Error performing DeleteObjects request: {}", source))]
     DeleteObjectsRequest { source: crate::client::retry::Error },
 
diff --git a/object_store/src/aws/dynamo.rs b/object_store/src/aws/dynamo.rs
index 2390187e7..9de67e5aa 100644
--- a/object_store/src/aws/dynamo.rs
+++ b/object_store/src/aws/dynamo.rs
@@ -451,18 +451,6 @@ struct PutItem<'a> {
     return_values_on_condition_check_failure: Option<ReturnValues>,
 }
 
-/// A DynamoDB [GetItem] payload
-///
-/// [GetItem]: https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_GetItem.html
-#[derive(Serialize)]
-#[serde(rename_all = "PascalCase")]
-struct GetItem<'a> {
-    /// The table name
-    table_name: &'a str,
-    /// The primary key
-    key: Map<'a, &'a str, AttributeValue<'a>>,
-}
-
 #[derive(Deserialize)]
 struct ErrorResponse<'a> {
     #[serde(rename = "__type")]
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 5bc6d56e7..f5204a536 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -409,14 +409,16 @@ impl MultipartStore for AmazonS3 {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{client::get::GetClient, tests::*};
+    use crate::client::get::GetClient;
+    use crate::integration::*;
+    use crate::tests::*;
     use hyper::HeaderMap;
 
     const NON_EXISTENT_NAME: &str = "nonexistentname";
 
     #[tokio::test]
     async fn s3_test() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = AmazonS3Builder::from_env();
 
         let integration = config.build().unwrap();
@@ -475,7 +477,7 @@ mod tests {
 
     #[tokio::test]
     async fn s3_test_get_nonexistent_location() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = AmazonS3Builder::from_env().build().unwrap();
 
         let location = Path::from_iter([NON_EXISTENT_NAME]);
@@ -488,7 +490,7 @@ mod tests {
 
     #[tokio::test]
     async fn s3_test_get_nonexistent_bucket() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = AmazonS3Builder::from_env().with_bucket_name(NON_EXISTENT_NAME);
         let integration = config.build().unwrap();
 
@@ -500,7 +502,7 @@ mod tests {
 
     #[tokio::test]
     async fn s3_test_put_nonexistent_bucket() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = AmazonS3Builder::from_env().with_bucket_name(NON_EXISTENT_NAME);
         let integration = config.build().unwrap();
 
@@ -513,7 +515,7 @@ mod tests {
 
     #[tokio::test]
     async fn s3_test_delete_nonexistent_location() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = AmazonS3Builder::from_env().build().unwrap();
 
         let location = Path::from_iter([NON_EXISTENT_NAME]);
@@ -523,7 +525,7 @@ mod tests {
 
     #[tokio::test]
     async fn s3_test_delete_nonexistent_bucket() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = AmazonS3Builder::from_env().with_bucket_name(NON_EXISTENT_NAME);
         let integration = config.build().unwrap();
 
@@ -560,7 +562,7 @@ mod tests {
     }
 
     async fn s3_encryption(store: &AmazonS3) {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
 
         let data = PutPayload::from(vec![3u8; 1024]);
 
diff --git a/object_store/src/azure/builder.rs b/object_store/src/azure/builder.rs
index ee0953456..c0c4e8983 100644
--- a/object_store/src/azure/builder.rs
+++ b/object_store/src/azure/builder.rs
@@ -89,11 +89,6 @@ enum Error {
 
     #[snafu(display("Configuration key: '{}' is not known.", key))]
     UnknownConfigurationKey { key: String },
-
-    #[snafu(display("Unable to extract metadata from headers: {}", source))]
-    Metadata {
-        source: crate::client::header::Error,
-    },
 }
 
 impl From<Error> for crate::Error {
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index 311bd72ff..be760c7dd 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -67,12 +67,6 @@ pub(crate) enum Error {
         path: String,
     },
 
-    #[snafu(display("Error getting get response body {}: {}", path, source))]
-    GetResponseBody {
-        source: reqwest::Error,
-        path: String,
-    },
-
     #[snafu(display("Error performing put request {}: {}", path, source))]
     PutRequest {
         source: crate::client::retry::Error,
@@ -94,11 +88,6 @@ pub(crate) enum Error {
     #[snafu(display("Got invalid list response: {}", source))]
     InvalidListResponse { source: quick_xml::de::DeError },
 
-    #[snafu(display("Error authorizing request: {}", source))]
-    Authorization {
-        source: crate::azure::credential::Error,
-    },
-
     #[snafu(display("Unable to extract metadata from headers: {}", source))]
     Metadata {
         source: crate::client::header::Error,
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index 755f3b126..f89a184f9 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -276,12 +276,13 @@ impl MultipartStore for MicrosoftAzure {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::integration::*;
     use crate::tests::*;
     use bytes::Bytes;
 
     #[tokio::test]
     async fn azure_blob_test() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = MicrosoftAzureBuilder::from_env().build().unwrap();
 
         put_get_delete_list(&integration).await;
diff --git a/object_store/src/chunked.rs b/object_store/src/chunked.rs
index a3bd76267..98cc20498 100644
--- a/object_store/src/chunked.rs
+++ b/object_store/src/chunked.rs
@@ -178,10 +178,10 @@ impl ObjectStore for ChunkedStore {
 mod tests {
     use futures::StreamExt;
 
+    use crate::integration::*;
     use crate::local::LocalFileSystem;
     use crate::memory::InMemory;
     use crate::path::Path;
-    use crate::tests::*;
 
     use super::*;
 
diff --git a/object_store/src/client/list.rs b/object_store/src/client/list.rs
index 371894dfe..2dbe20f3f 100644
--- a/object_store/src/client/list.rs
+++ b/object_store/src/client/list.rs
@@ -48,6 +48,7 @@ pub trait ListClientExt {
 
     fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>>;
 
+    #[allow(unused)]
     fn list_with_offset(
         &self,
         prefix: Option<&Path>,
diff --git a/object_store/src/gcp/builder.rs b/object_store/src/gcp/builder.rs
index e6da31207..82dab1443 100644
--- a/object_store/src/gcp/builder.rs
+++ b/object_store/src/gcp/builder.rs
@@ -60,11 +60,6 @@ enum Error {
     #[snafu(display("Configuration key: '{}' is not known.", key))]
     UnknownConfigurationKey { key: String },
 
-    #[snafu(display("Unable to extract metadata from headers: {}", source))]
-    Metadata {
-        source: crate::client::header::Error,
-    },
-
     #[snafu(display("GCP credential error: {}", source))]
     Credential { source: credential::Error },
 }
diff --git a/object_store/src/gcp/client.rs b/object_store/src/gcp/client.rs
index a54932565..35c64cc3d 100644
--- a/object_store/src/gcp/client.rs
+++ b/object_store/src/gcp/client.rs
@@ -81,12 +81,6 @@ enum Error {
     #[snafu(display("Got invalid put response: {}", source))]
     InvalidPutResponse { source: quick_xml::de::DeError },
 
-    #[snafu(display("Error performing post request {}: {}", path, source))]
-    PostRequest {
-        source: crate::client::retry::Error,
-        path: String,
-    },
-
     #[snafu(display("Unable to extract metadata from headers: {}", source))]
     Metadata {
         source: crate::client::header::Error,
diff --git a/object_store/src/gcp/credential.rs b/object_store/src/gcp/credential.rs
index d7fc2cea2..829db9b1a 100644
--- a/object_store/src/gcp/credential.rs
+++ b/object_store/src/gcp/credential.rs
@@ -536,7 +536,7 @@ impl ApplicationDefaultCredentials {
             let path = Path::new(&home).join(Self::CREDENTIALS_PATH);
 
             // It's expected for this file to not exist unless it has been explicitly configured by the user.
-            if path.try_exists().unwrap_or(false) {
+            if path.exists() {
                 return read_credentials_file::<Self>(path).map(Some);
             }
         }
diff --git a/object_store/src/gcp/mod.rs b/object_store/src/gcp/mod.rs
index 0ec6e7e82..039ec46b6 100644
--- a/object_store/src/gcp/mod.rs
+++ b/object_store/src/gcp/mod.rs
@@ -273,6 +273,7 @@ mod test {
 
     use credential::DEFAULT_GCS_BASE_URL;
 
+    use crate::integration::*;
     use crate::tests::*;
 
     use super::*;
@@ -281,7 +282,7 @@ mod test {
 
     #[tokio::test]
     async fn gcs_test() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = GoogleCloudStorageBuilder::from_env().build().unwrap();
 
         put_get_delete_list(&integration).await;
@@ -307,7 +308,7 @@ mod test {
     #[tokio::test]
     #[ignore]
     async fn gcs_test_sign() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = GoogleCloudStorageBuilder::from_env().build().unwrap();
 
         let client = reqwest::Client::new();
@@ -336,7 +337,7 @@ mod test {
 
     #[tokio::test]
     async fn gcs_test_get_nonexistent_location() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = GoogleCloudStorageBuilder::from_env().build().unwrap();
 
         let location = Path::from_iter([NON_EXISTENT_NAME]);
@@ -351,7 +352,7 @@ mod test {
 
     #[tokio::test]
     async fn gcs_test_get_nonexistent_bucket() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = GoogleCloudStorageBuilder::from_env();
         let integration = config.with_bucket_name(NON_EXISTENT_NAME).build().unwrap();
 
@@ -369,7 +370,7 @@ mod test {
 
     #[tokio::test]
     async fn gcs_test_delete_nonexistent_location() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let integration = GoogleCloudStorageBuilder::from_env().build().unwrap();
 
         let location = Path::from_iter([NON_EXISTENT_NAME]);
@@ -383,7 +384,7 @@ mod test {
 
     #[tokio::test]
     async fn gcs_test_delete_nonexistent_bucket() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = GoogleCloudStorageBuilder::from_env();
         let integration = config.with_bucket_name(NON_EXISTENT_NAME).build().unwrap();
 
@@ -398,7 +399,7 @@ mod test {
 
     #[tokio::test]
     async fn gcs_test_put_nonexistent_bucket() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let config = GoogleCloudStorageBuilder::from_env();
         let integration = config.with_bucket_name(NON_EXISTENT_NAME).build().unwrap();
 
diff --git a/object_store/src/http/mod.rs b/object_store/src/http/mod.rs
index 404211e57..4b1c927e7 100644
--- a/object_store/src/http/mod.rs
+++ b/object_store/src/http/mod.rs
@@ -64,9 +64,6 @@ enum Error {
     Metadata {
         source: crate::client::header::Error,
     },
-
-    #[snafu(display("Request error: {}", source))]
-    Reqwest { source: reqwest::Error },
 }
 
 impl From<Error> for crate::Error {
@@ -249,13 +246,14 @@ impl HttpBuilder {
 
 #[cfg(test)]
 mod tests {
+    use crate::integration::*;
     use crate::tests::*;
 
     use super::*;
 
     #[tokio::test]
     async fn http_test() {
-        crate::test_util::maybe_skip_integration!();
+        maybe_skip_integration!();
         let url = std::env::var("HTTP_URL").expect("HTTP_URL must be set");
         let options = ClientOptions::new().with_allow_http(true);
         let integration = HttpBuilder::new()
diff --git a/object_store/src/integration.rs b/object_store/src/integration.rs
new file mode 100644
index 000000000..9a7d11715
--- /dev/null
+++ b/object_store/src/integration.rs
@@ -0,0 +1,1082 @@
+// 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.
+
+//! Integration tests for custom object store implementations
+//!
+//! NB: These tests will delete everything present in the provided [`DynObjectStore`].
+//!
+//! These tests are not a stable part of the public API and breaking changes may be made
+//! in patch releases.
+//!
+//! They are intended solely for testing purposes.
+
+use crate::multipart::MultipartStore;
+use crate::path::Path;
+use crate::{
+    Attribute, Attributes, DynObjectStore, Error, GetOptions, GetRange, ObjectStore, PutMode,
+    PutPayload, UpdateVersion, WriteMultipart,
+};
+use bytes::Bytes;
+use futures::stream::FuturesUnordered;
+use futures::{StreamExt, TryStreamExt};
+use rand::distributions::Alphanumeric;
+use rand::{thread_rng, Rng};
+
+pub(crate) async fn flatten_list_stream(
+    storage: &DynObjectStore,
+    prefix: Option<&Path>,
+) -> crate::Result<Vec<Path>> {
+    storage
+        .list(prefix)
+        .map_ok(|meta| meta.location)
+        .try_collect::<Vec<Path>>()
+        .await
+}
+
+/// Tests basic read/write and listing operations
+pub async fn put_get_delete_list(storage: &DynObjectStore) {
+    delete_fixtures(storage).await;
+
+    let content_list = flatten_list_stream(storage, None).await.unwrap();
+    assert!(
+        content_list.is_empty(),
+        "Expected list to be empty; found: {content_list:?}"
+    );
+
+    let location = Path::from("test_dir/test_file.json");
+
+    let data = Bytes::from("arbitrary data");
+    storage.put(&location, data.clone().into()).await.unwrap();
+
+    let root = Path::from("/");
+
+    // List everything
+    let content_list = flatten_list_stream(storage, None).await.unwrap();
+    assert_eq!(content_list, &[location.clone()]);
+
+    // Should behave the same as no prefix
+    let content_list = flatten_list_stream(storage, Some(&root)).await.unwrap();
+    assert_eq!(content_list, &[location.clone()]);
+
+    // List with delimiter
+    let result = storage.list_with_delimiter(None).await.unwrap();
+    assert_eq!(&result.objects, &[]);
+    assert_eq!(result.common_prefixes.len(), 1);
+    assert_eq!(result.common_prefixes[0], Path::from("test_dir"));
+
+    // Should behave the same as no prefix
+    let result = storage.list_with_delimiter(Some(&root)).await.unwrap();
+    assert!(result.objects.is_empty());
+    assert_eq!(result.common_prefixes.len(), 1);
+    assert_eq!(result.common_prefixes[0], Path::from("test_dir"));
+
+    // Should return not found
+    let err = storage.get(&Path::from("test_dir")).await.unwrap_err();
+    assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
+
+    // Should return not found
+    let err = storage.head(&Path::from("test_dir")).await.unwrap_err();
+    assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
+
+    // List everything starting with a prefix that should return results
+    let prefix = Path::from("test_dir");
+    let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
+    assert_eq!(content_list, &[location.clone()]);
+
+    // List everything starting with a prefix that shouldn't return results
+    let prefix = Path::from("something");
+    let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
+    assert!(content_list.is_empty());
+
+    let read_data = storage.get(&location).await.unwrap().bytes().await.unwrap();
+    assert_eq!(&*read_data, data);
+
+    // Test range request
+    let range = 3..7;
+    let range_result = storage.get_range(&location, range.clone()).await;
+
+    let bytes = range_result.unwrap();
+    assert_eq!(bytes, data.slice(range.clone()));
+
+    let opts = GetOptions {
+        range: Some(GetRange::Bounded(2..5)),
+        ..Default::default()
+    };
+    let result = storage.get_opts(&location, opts).await.unwrap();
+    // Data is `"arbitrary data"`, length 14 bytes
+    assert_eq!(result.meta.size, 14); // Should return full object size (#5272)
+    assert_eq!(result.range, 2..5);
+    let bytes = result.bytes().await.unwrap();
+    assert_eq!(bytes, b"bit".as_ref());
+
+    let out_of_range = 200..300;
+    let out_of_range_result = storage.get_range(&location, out_of_range).await;
+
+    // Should be a non-fatal error
+    out_of_range_result.unwrap_err();
+
+    let opts = GetOptions {
+        range: Some(GetRange::Bounded(2..100)),
+        ..Default::default()
+    };
+    let result = storage.get_opts(&location, opts).await.unwrap();
+    assert_eq!(result.range, 2..14);
+    assert_eq!(result.meta.size, 14);
+    let bytes = result.bytes().await.unwrap();
+    assert_eq!(bytes, b"bitrary data".as_ref());
+
+    let opts = GetOptions {
+        range: Some(GetRange::Suffix(2)),
+        ..Default::default()
+    };
+    match storage.get_opts(&location, opts).await {
+        Ok(result) => {
+            assert_eq!(result.range, 12..14);
+            assert_eq!(result.meta.size, 14);
+            let bytes = result.bytes().await.unwrap();
+            assert_eq!(bytes, b"ta".as_ref());
+        }
+        Err(Error::NotSupported { .. }) => {}
+        Err(e) => panic!("{e}"),
+    }
+
+    let opts = GetOptions {
+        range: Some(GetRange::Suffix(100)),
+        ..Default::default()
+    };
+    match storage.get_opts(&location, opts).await {
+        Ok(result) => {
+            assert_eq!(result.range, 0..14);
+            assert_eq!(result.meta.size, 14);
+            let bytes = result.bytes().await.unwrap();
+            assert_eq!(bytes, b"arbitrary data".as_ref());
+        }
+        Err(Error::NotSupported { .. }) => {}
+        Err(e) => panic!("{e}"),
+    }
+
+    let opts = GetOptions {
+        range: Some(GetRange::Offset(3)),
+        ..Default::default()
+    };
+    let result = storage.get_opts(&location, opts).await.unwrap();
+    assert_eq!(result.range, 3..14);
+    assert_eq!(result.meta.size, 14);
+    let bytes = result.bytes().await.unwrap();
+    assert_eq!(bytes, b"itrary data".as_ref());
+
+    let opts = GetOptions {
+        range: Some(GetRange::Offset(100)),
+        ..Default::default()
+    };
+    storage.get_opts(&location, opts).await.unwrap_err();
+
+    let ranges = vec![0..1, 2..3, 0..5];
+    let bytes = storage.get_ranges(&location, &ranges).await.unwrap();
+    for (range, bytes) in ranges.iter().zip(bytes) {
+        assert_eq!(bytes, data.slice(range.clone()))
+    }
+
+    let head = storage.head(&location).await.unwrap();
+    assert_eq!(head.size, data.len());
+
+    storage.delete(&location).await.unwrap();
+
+    let content_list = flatten_list_stream(storage, None).await.unwrap();
+    assert!(content_list.is_empty());
+
+    let err = storage.get(&location).await.unwrap_err();
+    assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
+
+    let err = storage.head(&location).await.unwrap_err();
+    assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
+
+    // Test handling of paths containing an encoded delimiter
+
+    let file_with_delimiter = Path::from_iter(["a", "b/c", "foo.file"]);
+    storage
+        .put(&file_with_delimiter, "arbitrary".into())
+        .await
+        .unwrap();
+
+    let files = flatten_list_stream(storage, None).await.unwrap();
+    assert_eq!(files, vec![file_with_delimiter.clone()]);
+
+    let files = flatten_list_stream(storage, Some(&Path::from("a/b")))
+        .await
+        .unwrap();
+    assert!(files.is_empty());
+
+    let files = storage
+        .list_with_delimiter(Some(&Path::from("a/b")))
+        .await
+        .unwrap();
+    assert!(files.common_prefixes.is_empty());
+    assert!(files.objects.is_empty());
+
+    let files = storage
+        .list_with_delimiter(Some(&Path::from("a")))
+        .await
+        .unwrap();
+    assert_eq!(files.common_prefixes, vec![Path::from_iter(["a", "b/c"])]);
+    assert!(files.objects.is_empty());
+
+    let files = storage
+        .list_with_delimiter(Some(&Path::from_iter(["a", "b/c"])))
+        .await
+        .unwrap();
+    assert!(files.common_prefixes.is_empty());
+    assert_eq!(files.objects.len(), 1);
+    assert_eq!(files.objects[0].location, file_with_delimiter);
+
+    storage.delete(&file_with_delimiter).await.unwrap();
+
+    // Test handling of paths containing non-ASCII characters, e.g. emoji
+
+    let emoji_prefix = Path::from("🙀");
+    let emoji_file = Path::from("🙀/😀.parquet");
+    storage.put(&emoji_file, "arbitrary".into()).await.unwrap();
+
+    storage.head(&emoji_file).await.unwrap();
+    storage
+        .get(&emoji_file)
+        .await
+        .unwrap()
+        .bytes()
+        .await
+        .unwrap();
+
+    let files = flatten_list_stream(storage, Some(&emoji_prefix))
+        .await
+        .unwrap();
+
+    assert_eq!(files, vec![emoji_file.clone()]);
+
+    let dst = Path::from("foo.parquet");
+    storage.copy(&emoji_file, &dst).await.unwrap();
+    let mut files = flatten_list_stream(storage, None).await.unwrap();
+    files.sort_unstable();
+    assert_eq!(files, vec![emoji_file.clone(), dst.clone()]);
+
+    let dst2 = Path::from("new/nested/foo.parquet");
+    storage.copy(&emoji_file, &dst2).await.unwrap();
+    let mut files = flatten_list_stream(storage, None).await.unwrap();
+    files.sort_unstable();
+    assert_eq!(files, vec![emoji_file.clone(), dst.clone(), dst2.clone()]);
+
+    let dst3 = Path::from("new/nested2/bar.parquet");
+    storage.rename(&dst, &dst3).await.unwrap();
+    let mut files = flatten_list_stream(storage, None).await.unwrap();
+    files.sort_unstable();
+    assert_eq!(files, vec![emoji_file.clone(), dst2.clone(), dst3.clone()]);
+
+    let err = storage.head(&dst).await.unwrap_err();
+    assert!(matches!(err, Error::NotFound { .. }));
+
+    storage.delete(&emoji_file).await.unwrap();
+    storage.delete(&dst3).await.unwrap();
+    storage.delete(&dst2).await.unwrap();
+    let files = flatten_list_stream(storage, Some(&emoji_prefix))
+        .await
+        .unwrap();
+    assert!(files.is_empty());
+
+    // Test handling of paths containing percent-encoded sequences
+
+    // "HELLO" percent encoded
+    let hello_prefix = Path::parse("%48%45%4C%4C%4F").unwrap();
+    let path = hello_prefix.child("foo.parquet");
+
+    storage.put(&path, vec![0, 1].into()).await.unwrap();
+    let files = flatten_list_stream(storage, Some(&hello_prefix))
+        .await
+        .unwrap();
+    assert_eq!(files, vec![path.clone()]);
+
+    // Cannot list by decoded representation
+    let files = flatten_list_stream(storage, Some(&Path::from("HELLO")))
+        .await
+        .unwrap();
+    assert!(files.is_empty());
+
+    // Cannot access by decoded representation
+    let err = storage
+        .head(&Path::from("HELLO/foo.parquet"))
+        .await
+        .unwrap_err();
+    assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
+
+    storage.delete(&path).await.unwrap();
+
+    // Test handling of unicode paths
+    let path = Path::parse("🇦🇺/$shenanigans@@~.txt").unwrap();
+    storage.put(&path, "test".into()).await.unwrap();
+
+    let r = storage.get(&path).await.unwrap();
+    assert_eq!(r.bytes().await.unwrap(), "test");
+
+    let dir = Path::parse("🇦🇺").unwrap();
+    let r = storage.list_with_delimiter(None).await.unwrap();
+    assert!(r.common_prefixes.contains(&dir));
+
+    let r = storage.list_with_delimiter(Some(&dir)).await.unwrap();
+    assert_eq!(r.objects.len(), 1);
+    assert_eq!(r.objects[0].location, path);
+
+    storage.delete(&path).await.unwrap();
+
+    // Can also write non-percent encoded sequences
+    let path = Path::parse("%Q.parquet").unwrap();
+    storage.put(&path, vec![0, 1].into()).await.unwrap();
+
+    let files = flatten_list_stream(storage, None).await.unwrap();
+    assert_eq!(files, vec![path.clone()]);
+
+    storage.delete(&path).await.unwrap();
+
+    let path = Path::parse("foo bar/I contain spaces.parquet").unwrap();
+    storage.put(&path, vec![0, 1].into()).await.unwrap();
+    storage.head(&path).await.unwrap();
+
+    let files = flatten_list_stream(storage, Some(&Path::from("foo bar")))
+        .await
+        .unwrap();
+    assert_eq!(files, vec![path.clone()]);
+
+    storage.delete(&path).await.unwrap();
+
+    let files = flatten_list_stream(storage, None).await.unwrap();
+    assert!(files.is_empty(), "{files:?}");
+
+    // Test list order
+    let files = vec![
+        Path::from("a a/b.file"),
+        Path::parse("a%2Fa.file").unwrap(),
+        Path::from("a/😀.file"),
+        Path::from("a/a file"),
+        Path::parse("a/a%2F.file").unwrap(),
+        Path::from("a/a.file"),
+        Path::from("a/a/b.file"),
+        Path::from("a/b.file"),
+        Path::from("aa/a.file"),
+        Path::from("ab/a.file"),
+    ];
+
+    for file in &files {
+        storage.put(file, "foo".into()).await.unwrap();
+    }
+
+    let cases = [
+        (None, Path::from("a")),
+        (None, Path::from("a/a file")),
+        (None, Path::from("a/a/b.file")),
+        (None, Path::from("ab/a.file")),
+        (None, Path::from("a%2Fa.file")),
+        (None, Path::from("a/😀.file")),
+        (Some(Path::from("a")), Path::from("")),
+        (Some(Path::from("a")), Path::from("a")),
+        (Some(Path::from("a")), Path::from("a/😀")),
+        (Some(Path::from("a")), Path::from("a/😀.file")),
+        (Some(Path::from("a")), Path::from("a/b")),
+        (Some(Path::from("a")), Path::from("a/a/b.file")),
+    ];
+
+    for (prefix, offset) in cases {
+        let s = storage.list_with_offset(prefix.as_ref(), &offset);
+        let mut actual: Vec<_> = s.map_ok(|x| x.location).try_collect().await.unwrap();
+
+        actual.sort_unstable();
+
+        let expected: Vec<_> = files
+            .iter()
+            .filter(|x| {
+                let prefix_match = prefix.as_ref().map(|p| x.prefix_matches(p)).unwrap_or(true);
+                prefix_match && *x > &offset
+            })
+            .cloned()
+            .collect();
+
+        assert_eq!(actual, expected, "{prefix:?} - {offset:?}");
+    }
+
+    // Test bulk delete
+    let paths = vec![
+        Path::from("a/a.file"),
+        Path::from("a/a/b.file"),
+        Path::from("aa/a.file"),
+        Path::from("does_not_exist"),
+        Path::from("I'm a < & weird path"),
+        Path::from("ab/a.file"),
+        Path::from("a/😀.file"),
+    ];
+
+    storage.put(&paths[4], "foo".into()).await.unwrap();
+
+    let out_paths = storage
+        .delete_stream(futures::stream::iter(paths.clone()).map(Ok).boxed())
+        .collect::<Vec<_>>()
+        .await;
+
+    assert_eq!(out_paths.len(), paths.len());
+
+    let expect_errors = [3];
+
+    for (i, input_path) in paths.iter().enumerate() {
+        let err = storage.head(input_path).await.unwrap_err();
+        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
+
+        if expect_errors.contains(&i) {
+            // Some object stores will report NotFound, but others (such as S3) will
+            // report success regardless.
+            match &out_paths[i] {
+                Err(Error::NotFound { path: out_path, .. }) => {
+                    assert!(out_path.ends_with(&input_path.to_string()));
+                }
+                Ok(out_path) => {
+                    assert_eq!(out_path, input_path);
+                }
+                _ => panic!("unexpected error"),
+            }
+        } else {
+            assert_eq!(out_paths[i].as_ref().unwrap(), input_path);
+        }
+    }
+
+    delete_fixtures(storage).await;
+
+    let path = Path::from("empty");
+    storage.put(&path, PutPayload::default()).await.unwrap();
+    let meta = storage.head(&path).await.unwrap();
+    assert_eq!(meta.size, 0);
+    let data = storage.get(&path).await.unwrap().bytes().await.unwrap();
+    assert_eq!(data.len(), 0);
+
+    storage.delete(&path).await.unwrap();
+}
+
+/// Tests the ability to read and write [`Attributes`]
+pub async fn put_get_attributes(integration: &dyn ObjectStore) {
+    // Test handling of attributes
+    let attributes = Attributes::from_iter([
+        (Attribute::CacheControl, "max-age=604800"),
+        (
+            Attribute::ContentDisposition,
+            r#"attachment; filename="test.html""#,
+        ),
+        (Attribute::ContentEncoding, "gzip"),
+        (Attribute::ContentLanguage, "en-US"),
+        (Attribute::ContentType, "text/html; charset=utf-8"),
+    ]);
+
+    let path = Path::from("attributes");
+    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}"),
+    }
+}
+
+/// Tests conditional read requests
+pub async fn get_opts(storage: &dyn ObjectStore) {
+    let path = Path::from("test");
+    storage.put(&path, "foo".into()).await.unwrap();
+    let meta = storage.head(&path).await.unwrap();
+
+    let options = GetOptions {
+        if_unmodified_since: Some(meta.last_modified),
+        ..GetOptions::default()
+    };
+    match storage.get_opts(&path, options).await {
+        Ok(_) | Err(Error::NotSupported { .. }) => {}
+        Err(e) => panic!("{e}"),
+    }
+
+    let options = GetOptions {
+        if_unmodified_since: Some(meta.last_modified + chrono::Duration::try_hours(10).unwrap()),
+        ..GetOptions::default()
+    };
+    match storage.get_opts(&path, options).await {
+        Ok(_) | Err(Error::NotSupported { .. }) => {}
+        Err(e) => panic!("{e}"),
+    }
+
+    let options = GetOptions {
+        if_unmodified_since: Some(meta.last_modified - chrono::Duration::try_hours(10).unwrap()),
+        ..GetOptions::default()
+    };
+    match storage.get_opts(&path, options).await {
+        Err(Error::Precondition { .. } | Error::NotSupported { .. }) => {}
+        d => panic!("{d:?}"),
+    }
+
+    let options = GetOptions {
+        if_modified_since: Some(meta.last_modified),
+        ..GetOptions::default()
+    };
+    match storage.get_opts(&path, options).await {
+        Err(Error::NotModified { .. } | Error::NotSupported { .. }) => {}
+        d => panic!("{d:?}"),
+    }
+
+    let options = GetOptions {
+        if_modified_since: Some(meta.last_modified - chrono::Duration::try_hours(10).unwrap()),
+        ..GetOptions::default()
+    };
+    match storage.get_opts(&path, options).await {
+        Ok(_) | Err(Error::NotSupported { .. }) => {}
+        Err(e) => panic!("{e}"),
+    }
+
+    let tag = meta.e_tag.unwrap();
+    let options = GetOptions {
+        if_match: Some(tag.clone()),
+        ..GetOptions::default()
+    };
+    storage.get_opts(&path, options).await.unwrap();
+
+    let options = GetOptions {
+        if_match: Some("invalid".to_string()),
+        ..GetOptions::default()
+    };
+    let err = storage.get_opts(&path, options).await.unwrap_err();
+    assert!(matches!(err, Error::Precondition { .. }), "{err}");
+
+    let options = GetOptions {
+        if_none_match: Some(tag.clone()),
+        ..GetOptions::default()
+    };
+    let err = storage.get_opts(&path, options).await.unwrap_err();
+    assert!(matches!(err, Error::NotModified { .. }), "{err}");
+
+    let options = GetOptions {
+        if_none_match: Some("invalid".to_string()),
+        ..GetOptions::default()
+    };
+    storage.get_opts(&path, options).await.unwrap();
+
+    let result = storage.put(&path, "test".into()).await.unwrap();
+    let new_tag = result.e_tag.unwrap();
+    assert_ne!(tag, new_tag);
+
+    let meta = storage.head(&path).await.unwrap();
+    assert_eq!(meta.e_tag.unwrap(), new_tag);
+
+    let options = GetOptions {
+        if_match: Some(new_tag),
+        ..GetOptions::default()
+    };
+    storage.get_opts(&path, options).await.unwrap();
+
+    let options = GetOptions {
+        if_match: Some(tag),
+        ..GetOptions::default()
+    };
+    let err = storage.get_opts(&path, options).await.unwrap_err();
+    assert!(matches!(err, Error::Precondition { .. }), "{err}");
+
+    if let Some(version) = meta.version {
+        storage.put(&path, "bar".into()).await.unwrap();
+
+        let options = GetOptions {
+            version: Some(version),
+            ..GetOptions::default()
+        };
+
+        // Can retrieve previous version
+        let get_opts = storage.get_opts(&path, options).await.unwrap();
+        let old = get_opts.bytes().await.unwrap();
+        assert_eq!(old, b"test".as_slice());
+
+        // Current version contains the updated data
+        let current = storage.get(&path).await.unwrap().bytes().await.unwrap();
+        assert_eq!(&current, b"bar".as_slice());
+    }
+}
+
+/// Tests conditional writes
+pub async fn put_opts(storage: &dyn ObjectStore, supports_update: bool) {
+    // When using DynamoCommit repeated runs of this test will produce the same sequence of records in DynamoDB
+    // As a result each conditional operation will need to wait for the lease to timeout before proceeding
+    // One solution would be to clear DynamoDB before each test, but this would require non-trivial additional code
+    // so we instead just generate a random suffix for the filenames
+    let rng = thread_rng();
+    let suffix = String::from_utf8(rng.sample_iter(Alphanumeric).take(32).collect()).unwrap();
+
+    delete_fixtures(storage).await;
+    let path = Path::from(format!("put_opts_{suffix}"));
+    let v1 = storage
+        .put_opts(&path, "a".into(), PutMode::Create.into())
+        .await
+        .unwrap();
+
+    let err = storage
+        .put_opts(&path, "b".into(), PutMode::Create.into())
+        .await
+        .unwrap_err();
+    assert!(matches!(err, Error::AlreadyExists { .. }), "{err}");
+
+    let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
+    assert_eq!(b.as_ref(), b"a");
+
+    if !supports_update {
+        return;
+    }
+
+    let v2 = storage
+        .put_opts(&path, "c".into(), PutMode::Update(v1.clone().into()).into())
+        .await
+        .unwrap();
+
+    let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
+    assert_eq!(b.as_ref(), b"c");
+
+    let err = storage
+        .put_opts(&path, "d".into(), PutMode::Update(v1.into()).into())
+        .await
+        .unwrap_err();
+    assert!(matches!(err, Error::Precondition { .. }), "{err}");
+
+    storage
+        .put_opts(&path, "e".into(), PutMode::Update(v2.clone().into()).into())
+        .await
+        .unwrap();
+
+    let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
+    assert_eq!(b.as_ref(), b"e");
+
+    // Update not exists
+    let path = Path::from("I don't exist");
+    let err = storage
+        .put_opts(&path, "e".into(), PutMode::Update(v2.into()).into())
+        .await
+        .unwrap_err();
+    assert!(matches!(err, Error::Precondition { .. }), "{err}");
+
+    const NUM_WORKERS: usize = 5;
+    const NUM_INCREMENTS: usize = 10;
+
+    let path = Path::from(format!("RACE-{suffix}"));
+    let mut futures: FuturesUnordered<_> = (0..NUM_WORKERS)
+        .map(|_| async {
+            for _ in 0..NUM_INCREMENTS {
+                loop {
+                    match storage.get(&path).await {
+                        Ok(r) => {
+                            let mode = PutMode::Update(UpdateVersion {
+                                e_tag: r.meta.e_tag.clone(),
+                                version: r.meta.version.clone(),
+                            });
+
+                            let b = r.bytes().await.unwrap();
+                            let v: usize = std::str::from_utf8(&b).unwrap().parse().unwrap();
+                            let new = (v + 1).to_string();
+
+                            match storage.put_opts(&path, new.into(), mode.into()).await {
+                                Ok(_) => break,
+                                Err(Error::Precondition { .. }) => continue,
+                                Err(e) => return Err(e),
+                            }
+                        }
+                        Err(Error::NotFound { .. }) => {
+                            let mode = PutMode::Create;
+                            match storage.put_opts(&path, "1".into(), mode.into()).await {
+                                Ok(_) => break,
+                                Err(Error::AlreadyExists { .. }) => continue,
+                                Err(e) => return Err(e),
+                            }
+                        }
+                        Err(e) => return Err(e),
+                    }
+                }
+            }
+            Ok(())
+        })
+        .collect();
+
+    while futures.next().await.transpose().unwrap().is_some() {}
+    let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
+    let v = std::str::from_utf8(&b).unwrap().parse::<usize>().unwrap();
+    assert_eq!(v, NUM_WORKERS * NUM_INCREMENTS);
+}
+
+/// Returns a chunk of length `chunk_length`
+fn get_chunk(chunk_length: usize) -> Bytes {
+    let mut data = vec![0_u8; chunk_length];
+    let mut rng = thread_rng();
+    // Set a random selection of bytes
+    for _ in 0..1000 {
+        data[rng.gen_range(0..chunk_length)] = rng.gen();
+    }
+    data.into()
+}
+
+/// Returns `num_chunks` of length `chunks`
+fn get_chunks(chunk_length: usize, num_chunks: usize) -> Vec<Bytes> {
+    (0..num_chunks).map(|_| get_chunk(chunk_length)).collect()
+}
+
+/// Tests the ability to perform multipart writes
+pub async fn stream_get(storage: &DynObjectStore) {
+    let location = Path::from("test_dir/test_upload_file.txt");
+
+    // Can write to storage
+    let data = get_chunks(5 * 1024 * 1024, 3);
+    let bytes_expected = data.concat();
+    let mut upload = storage.put_multipart(&location).await.unwrap();
+    let uploads = data.into_iter().map(|x| upload.put_part(x.into()));
+    futures::future::try_join_all(uploads).await.unwrap();
+
+    // Object should not yet exist in store
+    let meta_res = storage.head(&location).await;
+    assert!(meta_res.is_err());
+    assert!(matches!(
+        meta_res.unwrap_err(),
+        crate::Error::NotFound { .. }
+    ));
+
+    let files = flatten_list_stream(storage, None).await.unwrap();
+    assert_eq!(&files, &[]);
+
+    let result = storage.list_with_delimiter(None).await.unwrap();
+    assert_eq!(&result.objects, &[]);
+
+    upload.complete().await.unwrap();
+
+    let bytes_written = storage.get(&location).await.unwrap().bytes().await.unwrap();
+    assert_eq!(bytes_expected, bytes_written);
+
+    // Can overwrite some storage
+    // Sizes chosen to ensure we write three parts
+    let data = get_chunks(3_200_000, 7);
+    let bytes_expected = data.concat();
+    let upload = storage.put_multipart(&location).await.unwrap();
+    let mut writer = WriteMultipart::new(upload);
+    for chunk in &data {
+        writer.write(chunk)
+    }
+    writer.finish().await.unwrap();
+    let bytes_written = storage.get(&location).await.unwrap().bytes().await.unwrap();
+    assert_eq!(bytes_expected, bytes_written);
+
+    // We can abort an empty write
+    let location = Path::from("test_dir/test_abort_upload.txt");
+    let mut upload = storage.put_multipart(&location).await.unwrap();
+    upload.abort().await.unwrap();
+    let get_res = storage.get(&location).await;
+    assert!(get_res.is_err());
+    assert!(matches!(
+        get_res.unwrap_err(),
+        crate::Error::NotFound { .. }
+    ));
+
+    // We can abort an in-progress write
+    let mut upload = storage.put_multipart(&location).await.unwrap();
+    upload
+        .put_part(data.first().unwrap().clone().into())
+        .await
+        .unwrap();
+
+    upload.abort().await.unwrap();
+    let get_res = storage.get(&location).await;
+    assert!(get_res.is_err());
+    assert!(matches!(get_res.unwrap_err(), Error::NotFound { .. }));
+}
+
+/// Tests that directories are transparent
+pub async fn list_uses_directories_correctly(storage: &DynObjectStore) {
+    delete_fixtures(storage).await;
+
+    let content_list = flatten_list_stream(storage, None).await.unwrap();
+    assert!(
+        content_list.is_empty(),
+        "Expected list to be empty; found: {content_list:?}"
+    );
+
+    let location1 = Path::from("foo/x.json");
+    let location2 = Path::from("foo.bar/y.json");
+
+    let data = PutPayload::from("arbitrary data");
+    storage.put(&location1, data.clone()).await.unwrap();
+    storage.put(&location2, data).await.unwrap();
+
+    let prefix = Path::from("foo");
+    let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
+    assert_eq!(content_list, &[location1.clone()]);
+
+    let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
+    assert_eq!(result.objects.len(), 1);
+    assert_eq!(result.objects[0].location, location1);
+    assert_eq!(result.common_prefixes, &[]);
+
+    // Listing an existing path (file) should return an empty list:
+    // https://github.com/apache/arrow-rs/issues/3712
+    let content_list = flatten_list_stream(storage, Some(&location1))
+        .await
+        .unwrap();
+    assert_eq!(content_list, &[]);
+
+    let list = storage.list_with_delimiter(Some(&location1)).await.unwrap();
+    assert_eq!(list.objects, &[]);
+    assert_eq!(list.common_prefixes, &[]);
+
+    let prefix = Path::from("foo/x");
+    let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
+    assert_eq!(content_list, &[]);
+
+    let list = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
+    assert_eq!(list.objects, &[]);
+    assert_eq!(list.common_prefixes, &[]);
+}
+
+/// Tests listing with delimiter
+pub async fn list_with_delimiter(storage: &DynObjectStore) {
+    delete_fixtures(storage).await;
+
+    // ==================== check: store is empty ====================
+    let content_list = flatten_list_stream(storage, None).await.unwrap();
+    assert!(content_list.is_empty());
+
+    // ==================== do: create files ====================
+    let data = Bytes::from("arbitrary data");
+
+    let files: Vec<_> = [
+        "test_file",
+        "mydb/wb/000/000/000.segment",
+        "mydb/wb/000/000/001.segment",
+        "mydb/wb/000/000/002.segment",
+        "mydb/wb/001/001/000.segment",
+        "mydb/wb/foo.json",
+        "mydb/wbwbwb/111/222/333.segment",
+        "mydb/data/whatevs",
+    ]
+    .iter()
+    .map(|&s| Path::from(s))
+    .collect();
+
+    for f in &files {
+        storage.put(f, data.clone().into()).await.unwrap();
+    }
+
+    // ==================== check: prefix-list `mydb/wb` (directory) ====================
+    let prefix = Path::from("mydb/wb");
+
+    let expected_000 = Path::from("mydb/wb/000");
+    let expected_001 = Path::from("mydb/wb/001");
+    let expected_location = Path::from("mydb/wb/foo.json");
+
+    let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
+
+    assert_eq!(result.common_prefixes, vec![expected_000, expected_001]);
+    assert_eq!(result.objects.len(), 1);
+
+    let object = &result.objects[0];
+
+    assert_eq!(object.location, expected_location);
+    assert_eq!(object.size, data.len());
+
+    // ==================== check: prefix-list `mydb/wb/000/000/001` (partial filename doesn't match) ====================
+    let prefix = Path::from("mydb/wb/000/000/001");
+
+    let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
+    assert!(result.common_prefixes.is_empty());
+    assert_eq!(result.objects.len(), 0);
+
+    // ==================== check: prefix-list `not_there` (non-existing prefix) ====================
+    let prefix = Path::from("not_there");
+
+    let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
+    assert!(result.common_prefixes.is_empty());
+    assert!(result.objects.is_empty());
+
+    // ==================== do: remove all files ====================
+    for f in &files {
+        storage.delete(f).await.unwrap();
+    }
+
+    // ==================== check: store is empty ====================
+    let content_list = flatten_list_stream(storage, None).await.unwrap();
+    assert!(content_list.is_empty());
+}
+
+/// Tests fetching a non-existent object returns a not found error
+pub async fn get_nonexistent_object(
+    storage: &DynObjectStore,
+    location: Option<Path>,
+) -> crate::Result<Bytes> {
+    let location = location.unwrap_or_else(|| Path::from("this_file_should_not_exist"));
+
+    let err = storage.head(&location).await.unwrap_err();
+    assert!(matches!(err, Error::NotFound { .. }));
+
+    storage.get(&location).await?.bytes().await
+}
+
+/// Tests copying
+pub async fn rename_and_copy(storage: &DynObjectStore) {
+    // Create two objects
+    let path1 = Path::from("test1");
+    let path2 = Path::from("test2");
+    let contents1 = Bytes::from("cats");
+    let contents2 = Bytes::from("dogs");
+
+    // copy() make both objects identical
+    storage.put(&path1, contents1.clone().into()).await.unwrap();
+    storage.put(&path2, contents2.clone().into()).await.unwrap();
+    storage.copy(&path1, &path2).await.unwrap();
+    let new_contents = storage.get(&path2).await.unwrap().bytes().await.unwrap();
+    assert_eq!(&new_contents, &contents1);
+
+    // rename() copies contents and deletes original
+    storage.put(&path1, contents1.clone().into()).await.unwrap();
+    storage.put(&path2, contents2.clone().into()).await.unwrap();
+    storage.rename(&path1, &path2).await.unwrap();
+    let new_contents = storage.get(&path2).await.unwrap().bytes().await.unwrap();
+    assert_eq!(&new_contents, &contents1);
+    let result = storage.get(&path1).await;
+    assert!(result.is_err());
+    assert!(matches!(result.unwrap_err(), Error::NotFound { .. }));
+
+    // Clean up
+    storage.delete(&path2).await.unwrap();
+}
+
+/// Tests copy if not exists
+pub async fn copy_if_not_exists(storage: &DynObjectStore) {
+    // Create two objects
+    let path1 = Path::from("test1");
+    let path2 = Path::from("not_exists_nested/test2");
+    let contents1 = Bytes::from("cats");
+    let contents2 = Bytes::from("dogs");
+
+    // copy_if_not_exists() errors if destination already exists
+    storage.put(&path1, contents1.clone().into()).await.unwrap();
+    storage.put(&path2, contents2.clone().into()).await.unwrap();
+    let result = storage.copy_if_not_exists(&path1, &path2).await;
+    assert!(result.is_err());
+    assert!(matches!(
+        result.unwrap_err(),
+        crate::Error::AlreadyExists { .. }
+    ));
+
+    // copy_if_not_exists() copies contents and allows deleting original
+    storage.delete(&path2).await.unwrap();
+    storage.copy_if_not_exists(&path1, &path2).await.unwrap();
+    storage.delete(&path1).await.unwrap();
+    let new_contents = storage.get(&path2).await.unwrap().bytes().await.unwrap();
+    assert_eq!(&new_contents, &contents1);
+    let result = storage.get(&path1).await;
+    assert!(result.is_err());
+    assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
+
+    // Clean up
+    storage.delete(&path2).await.unwrap();
+}
+
+/// Tests copy and renaming behaviour of non-existent objects
+pub async fn copy_rename_nonexistent_object(storage: &DynObjectStore) {
+    // Create empty source object
+    let path1 = Path::from("test1");
+
+    // Create destination object
+    let path2 = Path::from("test2");
+    storage.put(&path2, "hello".into()).await.unwrap();
+
+    // copy() errors if source does not exist
+    let result = storage.copy(&path1, &path2).await;
+    assert!(result.is_err());
+    assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
+
+    // rename() errors if source does not exist
+    let result = storage.rename(&path1, &path2).await;
+    assert!(result.is_err());
+    assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
+
+    // copy_if_not_exists() errors if source does not exist
+    let result = storage.copy_if_not_exists(&path1, &path2).await;
+    assert!(result.is_err());
+    assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
+
+    // Clean up
+    storage.delete(&path2).await.unwrap();
+}
+
+/// Tests [`MultipartStore`]
+pub async fn multipart(storage: &dyn ObjectStore, multipart: &dyn MultipartStore) {
+    let path = Path::from("test_multipart");
+    let chunk_size = 5 * 1024 * 1024;
+
+    let chunks = get_chunks(chunk_size, 2);
+
+    let id = multipart.create_multipart(&path).await.unwrap();
+
+    let parts: Vec<_> = futures::stream::iter(chunks)
+        .enumerate()
+        .map(|(idx, b)| multipart.put_part(&path, &id, idx, b.into()))
+        .buffered(2)
+        .try_collect()
+        .await
+        .unwrap();
+
+    multipart
+        .complete_multipart(&path, &id, parts)
+        .await
+        .unwrap();
+
+    let meta = storage.head(&path).await.unwrap();
+    assert_eq!(meta.size, chunk_size * 2);
+
+    // Empty case
+    let path = Path::from("test_empty_multipart");
+
+    let id = multipart.create_multipart(&path).await.unwrap();
+
+    let parts = vec![];
+
+    multipart
+        .complete_multipart(&path, &id, parts)
+        .await
+        .unwrap();
+
+    let meta = storage.head(&path).await.unwrap();
+    assert_eq!(meta.size, 0);
+}
+
+async fn delete_fixtures(storage: &DynObjectStore) {
+    let paths = storage.list(None).map_ok(|meta| meta.location).boxed();
+    storage
+        .delete_stream(paths)
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap();
+}
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 9a8f77b4d..bdf870f45 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -545,6 +545,9 @@ mod util;
 
 mod attributes;
 
+#[cfg(any(feature = "integration", test))]
+pub mod integration;
+
 pub use attributes::*;
 
 pub use parse::{parse_url, parse_url_opts};
@@ -1285,9 +1288,11 @@ impl From<Error> for std::io::Error {
 }
 
 #[cfg(test)]
-mod test_util {
+mod tests {
     use super::*;
-    use futures::TryStreamExt;
+    use crate::buffered::BufWriter;
+    use chrono::TimeZone;
+    use tokio::io::AsyncWriteExt;
 
     macro_rules! maybe_skip_integration {
         () => {
@@ -1299,1054 +1304,19 @@ mod test_util {
     }
     pub(crate) use maybe_skip_integration;
 
-    pub async fn flatten_list_stream(
-        storage: &DynObjectStore,
-        prefix: Option<&Path>,
-    ) -> Result<Vec<Path>> {
-        storage
-            .list(prefix)
-            .map_ok(|meta| meta.location)
-            .try_collect::<Vec<Path>>()
-            .await
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::buffered::BufWriter;
-    use crate::multipart::MultipartStore;
-    use crate::test_util::flatten_list_stream;
-    use chrono::TimeZone;
-    use futures::stream::FuturesUnordered;
-    use rand::distributions::Alphanumeric;
-    use rand::{thread_rng, Rng};
-    use tokio::io::AsyncWriteExt;
-
-    pub(crate) async fn put_get_delete_list(storage: &DynObjectStore) {
-        delete_fixtures(storage).await;
-
-        let content_list = flatten_list_stream(storage, None).await.unwrap();
-        assert!(
-            content_list.is_empty(),
-            "Expected list to be empty; found: {content_list:?}"
-        );
-
-        let location = Path::from("test_dir/test_file.json");
-
-        let data = Bytes::from("arbitrary data");
-        storage.put(&location, data.clone().into()).await.unwrap();
-
-        let root = Path::from("/");
-
-        // List everything
-        let content_list = flatten_list_stream(storage, None).await.unwrap();
-        assert_eq!(content_list, &[location.clone()]);
-
-        // Should behave the same as no prefix
-        let content_list = flatten_list_stream(storage, Some(&root)).await.unwrap();
-        assert_eq!(content_list, &[location.clone()]);
-
-        // List with delimiter
-        let result = storage.list_with_delimiter(None).await.unwrap();
-        assert_eq!(&result.objects, &[]);
-        assert_eq!(result.common_prefixes.len(), 1);
-        assert_eq!(result.common_prefixes[0], Path::from("test_dir"));
-
-        // Should behave the same as no prefix
-        let result = storage.list_with_delimiter(Some(&root)).await.unwrap();
-        assert!(result.objects.is_empty());
-        assert_eq!(result.common_prefixes.len(), 1);
-        assert_eq!(result.common_prefixes[0], Path::from("test_dir"));
-
-        // Should return not found
-        let err = storage.get(&Path::from("test_dir")).await.unwrap_err();
-        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
-
-        // Should return not found
-        let err = storage.head(&Path::from("test_dir")).await.unwrap_err();
-        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
-
-        // List everything starting with a prefix that should return results
-        let prefix = Path::from("test_dir");
-        let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
-        assert_eq!(content_list, &[location.clone()]);
-
-        // List everything starting with a prefix that shouldn't return results
-        let prefix = Path::from("something");
-        let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
-        assert!(content_list.is_empty());
-
-        let read_data = storage.get(&location).await.unwrap().bytes().await.unwrap();
-        assert_eq!(&*read_data, data);
-
-        // Test range request
-        let range = 3..7;
-        let range_result = storage.get_range(&location, range.clone()).await;
-
-        let bytes = range_result.unwrap();
-        assert_eq!(bytes, data.slice(range.clone()));
-
-        let opts = GetOptions {
-            range: Some(GetRange::Bounded(2..5)),
-            ..Default::default()
-        };
-        let result = storage.get_opts(&location, opts).await.unwrap();
-        // Data is `"arbitrary data"`, length 14 bytes
-        assert_eq!(result.meta.size, 14); // Should return full object size (#5272)
-        assert_eq!(result.range, 2..5);
-        let bytes = result.bytes().await.unwrap();
-        assert_eq!(bytes, b"bit".as_ref());
-
-        let out_of_range = 200..300;
-        let out_of_range_result = storage.get_range(&location, out_of_range).await;
-
-        // Should be a non-fatal error
-        out_of_range_result.unwrap_err();
-
-        let opts = GetOptions {
-            range: Some(GetRange::Bounded(2..100)),
-            ..Default::default()
-        };
-        let result = storage.get_opts(&location, opts).await.unwrap();
-        assert_eq!(result.range, 2..14);
-        assert_eq!(result.meta.size, 14);
-        let bytes = result.bytes().await.unwrap();
-        assert_eq!(bytes, b"bitrary data".as_ref());
-
-        let opts = GetOptions {
-            range: Some(GetRange::Suffix(2)),
-            ..Default::default()
-        };
-        match storage.get_opts(&location, opts).await {
-            Ok(result) => {
-                assert_eq!(result.range, 12..14);
-                assert_eq!(result.meta.size, 14);
-                let bytes = result.bytes().await.unwrap();
-                assert_eq!(bytes, b"ta".as_ref());
-            }
-            Err(Error::NotSupported { .. }) => {}
-            Err(e) => panic!("{e}"),
-        }
-
-        let opts = GetOptions {
-            range: Some(GetRange::Suffix(100)),
-            ..Default::default()
-        };
-        match storage.get_opts(&location, opts).await {
-            Ok(result) => {
-                assert_eq!(result.range, 0..14);
-                assert_eq!(result.meta.size, 14);
-                let bytes = result.bytes().await.unwrap();
-                assert_eq!(bytes, b"arbitrary data".as_ref());
-            }
-            Err(Error::NotSupported { .. }) => {}
-            Err(e) => panic!("{e}"),
-        }
-
-        let opts = GetOptions {
-            range: Some(GetRange::Offset(3)),
-            ..Default::default()
-        };
-        let result = storage.get_opts(&location, opts).await.unwrap();
-        assert_eq!(result.range, 3..14);
-        assert_eq!(result.meta.size, 14);
-        let bytes = result.bytes().await.unwrap();
-        assert_eq!(bytes, b"itrary data".as_ref());
-
-        let opts = GetOptions {
-            range: Some(GetRange::Offset(100)),
-            ..Default::default()
-        };
-        storage.get_opts(&location, opts).await.unwrap_err();
-
-        let ranges = vec![0..1, 2..3, 0..5];
-        let bytes = storage.get_ranges(&location, &ranges).await.unwrap();
-        for (range, bytes) in ranges.iter().zip(bytes) {
-            assert_eq!(bytes, data.slice(range.clone()))
-        }
-
-        let head = storage.head(&location).await.unwrap();
-        assert_eq!(head.size, data.len());
-
-        storage.delete(&location).await.unwrap();
-
-        let content_list = flatten_list_stream(storage, None).await.unwrap();
-        assert!(content_list.is_empty());
-
-        let err = storage.get(&location).await.unwrap_err();
-        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
-
-        let err = storage.head(&location).await.unwrap_err();
-        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
-
-        // Test handling of paths containing an encoded delimiter
-
-        let file_with_delimiter = Path::from_iter(["a", "b/c", "foo.file"]);
-        storage
-            .put(&file_with_delimiter, "arbitrary".into())
-            .await
-            .unwrap();
-
-        let files = flatten_list_stream(storage, None).await.unwrap();
-        assert_eq!(files, vec![file_with_delimiter.clone()]);
-
-        let files = flatten_list_stream(storage, Some(&Path::from("a/b")))
-            .await
-            .unwrap();
-        assert!(files.is_empty());
-
-        let files = storage
-            .list_with_delimiter(Some(&Path::from("a/b")))
-            .await
-            .unwrap();
-        assert!(files.common_prefixes.is_empty());
-        assert!(files.objects.is_empty());
-
-        let files = storage
-            .list_with_delimiter(Some(&Path::from("a")))
-            .await
-            .unwrap();
-        assert_eq!(files.common_prefixes, vec![Path::from_iter(["a", "b/c"])]);
-        assert!(files.objects.is_empty());
-
-        let files = storage
-            .list_with_delimiter(Some(&Path::from_iter(["a", "b/c"])))
-            .await
-            .unwrap();
-        assert!(files.common_prefixes.is_empty());
-        assert_eq!(files.objects.len(), 1);
-        assert_eq!(files.objects[0].location, file_with_delimiter);
-
-        storage.delete(&file_with_delimiter).await.unwrap();
-
-        // Test handling of paths containing non-ASCII characters, e.g. emoji
-
-        let emoji_prefix = Path::from("🙀");
-        let emoji_file = Path::from("🙀/😀.parquet");
-        storage.put(&emoji_file, "arbitrary".into()).await.unwrap();
-
-        storage.head(&emoji_file).await.unwrap();
-        storage
-            .get(&emoji_file)
-            .await
-            .unwrap()
-            .bytes()
-            .await
-            .unwrap();
-
-        let files = flatten_list_stream(storage, Some(&emoji_prefix))
-            .await
-            .unwrap();
-
-        assert_eq!(files, vec![emoji_file.clone()]);
-
-        let dst = Path::from("foo.parquet");
-        storage.copy(&emoji_file, &dst).await.unwrap();
-        let mut files = flatten_list_stream(storage, None).await.unwrap();
-        files.sort_unstable();
-        assert_eq!(files, vec![emoji_file.clone(), dst.clone()]);
-
-        let dst2 = Path::from("new/nested/foo.parquet");
-        storage.copy(&emoji_file, &dst2).await.unwrap();
-        let mut files = flatten_list_stream(storage, None).await.unwrap();
-        files.sort_unstable();
-        assert_eq!(files, vec![emoji_file.clone(), dst.clone(), dst2.clone()]);
-
-        let dst3 = Path::from("new/nested2/bar.parquet");
-        storage.rename(&dst, &dst3).await.unwrap();
-        let mut files = flatten_list_stream(storage, None).await.unwrap();
-        files.sort_unstable();
-        assert_eq!(files, vec![emoji_file.clone(), dst2.clone(), dst3.clone()]);
-
-        let err = storage.head(&dst).await.unwrap_err();
-        assert!(matches!(err, Error::NotFound { .. }));
-
-        storage.delete(&emoji_file).await.unwrap();
-        storage.delete(&dst3).await.unwrap();
-        storage.delete(&dst2).await.unwrap();
-        let files = flatten_list_stream(storage, Some(&emoji_prefix))
-            .await
-            .unwrap();
-        assert!(files.is_empty());
-
-        // Test handling of paths containing percent-encoded sequences
-
-        // "HELLO" percent encoded
-        let hello_prefix = Path::parse("%48%45%4C%4C%4F").unwrap();
-        let path = hello_prefix.child("foo.parquet");
-
-        storage.put(&path, vec![0, 1].into()).await.unwrap();
-        let files = flatten_list_stream(storage, Some(&hello_prefix))
-            .await
-            .unwrap();
-        assert_eq!(files, vec![path.clone()]);
-
-        // Cannot list by decoded representation
-        let files = flatten_list_stream(storage, Some(&Path::from("HELLO")))
-            .await
-            .unwrap();
-        assert!(files.is_empty());
-
-        // Cannot access by decoded representation
-        let err = storage
-            .head(&Path::from("HELLO/foo.parquet"))
-            .await
-            .unwrap_err();
-        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
-
-        storage.delete(&path).await.unwrap();
-
-        // Test handling of unicode paths
-        let path = Path::parse("🇦🇺/$shenanigans@@~.txt").unwrap();
-        storage.put(&path, "test".into()).await.unwrap();
-
-        let r = storage.get(&path).await.unwrap();
-        assert_eq!(r.bytes().await.unwrap(), "test");
-
-        let dir = Path::parse("🇦🇺").unwrap();
-        let r = storage.list_with_delimiter(None).await.unwrap();
-        assert!(r.common_prefixes.contains(&dir));
-
-        let r = storage.list_with_delimiter(Some(&dir)).await.unwrap();
-        assert_eq!(r.objects.len(), 1);
-        assert_eq!(r.objects[0].location, path);
-
-        storage.delete(&path).await.unwrap();
-
-        // Can also write non-percent encoded sequences
-        let path = Path::parse("%Q.parquet").unwrap();
-        storage.put(&path, vec![0, 1].into()).await.unwrap();
-
-        let files = flatten_list_stream(storage, None).await.unwrap();
-        assert_eq!(files, vec![path.clone()]);
-
-        storage.delete(&path).await.unwrap();
-
-        let path = Path::parse("foo bar/I contain spaces.parquet").unwrap();
-        storage.put(&path, vec![0, 1].into()).await.unwrap();
-        storage.head(&path).await.unwrap();
-
-        let files = flatten_list_stream(storage, Some(&Path::from("foo bar")))
-            .await
-            .unwrap();
-        assert_eq!(files, vec![path.clone()]);
-
-        storage.delete(&path).await.unwrap();
-
-        let files = flatten_list_stream(storage, None).await.unwrap();
-        assert!(files.is_empty(), "{files:?}");
-
-        // Test list order
-        let files = vec![
-            Path::from("a a/b.file"),
-            Path::parse("a%2Fa.file").unwrap(),
-            Path::from("a/😀.file"),
-            Path::from("a/a file"),
-            Path::parse("a/a%2F.file").unwrap(),
-            Path::from("a/a.file"),
-            Path::from("a/a/b.file"),
-            Path::from("a/b.file"),
-            Path::from("aa/a.file"),
-            Path::from("ab/a.file"),
-        ];
-
-        for file in &files {
-            storage.put(file, "foo".into()).await.unwrap();
-        }
-
-        let cases = [
-            (None, Path::from("a")),
-            (None, Path::from("a/a file")),
-            (None, Path::from("a/a/b.file")),
-            (None, Path::from("ab/a.file")),
-            (None, Path::from("a%2Fa.file")),
-            (None, Path::from("a/😀.file")),
-            (Some(Path::from("a")), Path::from("")),
-            (Some(Path::from("a")), Path::from("a")),
-            (Some(Path::from("a")), Path::from("a/😀")),
-            (Some(Path::from("a")), Path::from("a/😀.file")),
-            (Some(Path::from("a")), Path::from("a/b")),
-            (Some(Path::from("a")), Path::from("a/a/b.file")),
-        ];
-
-        for (prefix, offset) in cases {
-            let s = storage.list_with_offset(prefix.as_ref(), &offset);
-            let mut actual: Vec<_> = s.map_ok(|x| x.location).try_collect().await.unwrap();
-
-            actual.sort_unstable();
-
-            let expected: Vec<_> = files
-                .iter()
-                .filter(|x| {
-                    let prefix_match = prefix.as_ref().map(|p| x.prefix_matches(p)).unwrap_or(true);
-                    prefix_match && *x > &offset
-                })
-                .cloned()
-                .collect();
-
-            assert_eq!(actual, expected, "{prefix:?} - {offset:?}");
-        }
-
-        // Test bulk delete
-        let paths = vec![
-            Path::from("a/a.file"),
-            Path::from("a/a/b.file"),
-            Path::from("aa/a.file"),
-            Path::from("does_not_exist"),
-            Path::from("I'm a < & weird path"),
-            Path::from("ab/a.file"),
-            Path::from("a/😀.file"),
-        ];
-
-        storage.put(&paths[4], "foo".into()).await.unwrap();
-
-        let out_paths = storage
-            .delete_stream(futures::stream::iter(paths.clone()).map(Ok).boxed())
-            .collect::<Vec<_>>()
-            .await;
-
-        assert_eq!(out_paths.len(), paths.len());
-
-        let expect_errors = [3];
-
-        for (i, input_path) in paths.iter().enumerate() {
-            let err = storage.head(input_path).await.unwrap_err();
-            assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
-
-            if expect_errors.contains(&i) {
-                // Some object stores will report NotFound, but others (such as S3) will
-                // report success regardless.
-                match &out_paths[i] {
-                    Err(Error::NotFound { path: out_path, .. }) => {
-                        assert!(out_path.ends_with(&input_path.to_string()));
-                    }
-                    Ok(out_path) => {
-                        assert_eq!(out_path, input_path);
-                    }
-                    _ => panic!("unexpected error"),
-                }
-            } else {
-                assert_eq!(out_paths[i].as_ref().unwrap(), input_path);
-            }
-        }
-
-        delete_fixtures(storage).await;
-
-        let path = Path::from("empty");
-        storage.put(&path, PutPayload::default()).await.unwrap();
-        let meta = storage.head(&path).await.unwrap();
-        assert_eq!(meta.size, 0);
-        let data = storage.get(&path).await.unwrap().bytes().await.unwrap();
-        assert_eq!(data.len(), 0);
-
-        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::CacheControl, "max-age=604800"),
-            (
-                Attribute::ContentDisposition,
-                r#"attachment; filename="test.html""#,
-            ),
-            (Attribute::ContentEncoding, "gzip"),
-            (Attribute::ContentLanguage, "en-US"),
-            (Attribute::ContentType, "text/html; charset=utf-8"),
-        ]);
-
-        let path = Path::from("attributes");
-        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();
-        let meta = storage.head(&path).await.unwrap();
-
-        let options = GetOptions {
-            if_unmodified_since: Some(meta.last_modified),
-            ..GetOptions::default()
-        };
-        match storage.get_opts(&path, options).await {
-            Ok(_) | Err(Error::NotSupported { .. }) => {}
-            Err(e) => panic!("{e}"),
-        }
-
-        let options = GetOptions {
-            if_unmodified_since: Some(
-                meta.last_modified + chrono::Duration::try_hours(10).unwrap(),
-            ),
-            ..GetOptions::default()
-        };
-        match storage.get_opts(&path, options).await {
-            Ok(_) | Err(Error::NotSupported { .. }) => {}
-            Err(e) => panic!("{e}"),
-        }
-
-        let options = GetOptions {
-            if_unmodified_since: Some(
-                meta.last_modified - chrono::Duration::try_hours(10).unwrap(),
-            ),
-            ..GetOptions::default()
-        };
-        match storage.get_opts(&path, options).await {
-            Err(Error::Precondition { .. } | Error::NotSupported { .. }) => {}
-            d => panic!("{d:?}"),
-        }
-
-        let options = GetOptions {
-            if_modified_since: Some(meta.last_modified),
-            ..GetOptions::default()
-        };
-        match storage.get_opts(&path, options).await {
-            Err(Error::NotModified { .. } | Error::NotSupported { .. }) => {}
-            d => panic!("{d:?}"),
-        }
-
-        let options = GetOptions {
-            if_modified_since: Some(meta.last_modified - chrono::Duration::try_hours(10).unwrap()),
-            ..GetOptions::default()
-        };
-        match storage.get_opts(&path, options).await {
-            Ok(_) | Err(Error::NotSupported { .. }) => {}
-            Err(e) => panic!("{e}"),
-        }
-
-        let tag = meta.e_tag.unwrap();
-        let options = GetOptions {
-            if_match: Some(tag.clone()),
-            ..GetOptions::default()
-        };
-        storage.get_opts(&path, options).await.unwrap();
-
-        let options = GetOptions {
-            if_match: Some("invalid".to_string()),
-            ..GetOptions::default()
-        };
-        let err = storage.get_opts(&path, options).await.unwrap_err();
-        assert!(matches!(err, Error::Precondition { .. }), "{err}");
-
-        let options = GetOptions {
-            if_none_match: Some(tag.clone()),
-            ..GetOptions::default()
-        };
-        let err = storage.get_opts(&path, options).await.unwrap_err();
-        assert!(matches!(err, Error::NotModified { .. }), "{err}");
-
-        let options = GetOptions {
-            if_none_match: Some("invalid".to_string()),
-            ..GetOptions::default()
-        };
-        storage.get_opts(&path, options).await.unwrap();
-
-        let result = storage.put(&path, "test".into()).await.unwrap();
-        let new_tag = result.e_tag.unwrap();
-        assert_ne!(tag, new_tag);
-
-        let meta = storage.head(&path).await.unwrap();
-        assert_eq!(meta.e_tag.unwrap(), new_tag);
-
-        let options = GetOptions {
-            if_match: Some(new_tag),
-            ..GetOptions::default()
-        };
-        storage.get_opts(&path, options).await.unwrap();
-
-        let options = GetOptions {
-            if_match: Some(tag),
-            ..GetOptions::default()
-        };
-        let err = storage.get_opts(&path, options).await.unwrap_err();
-        assert!(matches!(err, Error::Precondition { .. }), "{err}");
-
-        if let Some(version) = meta.version {
-            storage.put(&path, "bar".into()).await.unwrap();
-
-            let options = GetOptions {
-                version: Some(version),
-                ..GetOptions::default()
-            };
-
-            // Can retrieve previous version
-            let get_opts = storage.get_opts(&path, options).await.unwrap();
-            let old = get_opts.bytes().await.unwrap();
-            assert_eq!(old, b"test".as_slice());
-
-            // Current version contains the updated data
-            let current = storage.get(&path).await.unwrap().bytes().await.unwrap();
-            assert_eq!(&current, b"bar".as_slice());
-        }
-    }
-
-    pub(crate) async fn put_opts(storage: &dyn ObjectStore, supports_update: bool) {
-        // When using DynamoCommit repeated runs of this test will produce the same sequence of records in DynamoDB
-        // As a result each conditional operation will need to wait for the lease to timeout before proceeding
-        // One solution would be to clear DynamoDB before each test, but this would require non-trivial additional code
-        // so we instead just generate a random suffix for the filenames
-        let rng = thread_rng();
-        let suffix = String::from_utf8(rng.sample_iter(Alphanumeric).take(32).collect()).unwrap();
-
-        delete_fixtures(storage).await;
-        let path = Path::from(format!("put_opts_{suffix}"));
-        let v1 = storage
-            .put_opts(&path, "a".into(), PutMode::Create.into())
-            .await
-            .unwrap();
-
-        let err = storage
-            .put_opts(&path, "b".into(), PutMode::Create.into())
-            .await
-            .unwrap_err();
-        assert!(matches!(err, Error::AlreadyExists { .. }), "{err}");
-
-        let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
-        assert_eq!(b.as_ref(), b"a");
-
-        if !supports_update {
-            return;
-        }
-
-        let v2 = storage
-            .put_opts(&path, "c".into(), PutMode::Update(v1.clone().into()).into())
-            .await
-            .unwrap();
-
-        let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
-        assert_eq!(b.as_ref(), b"c");
-
-        let err = storage
-            .put_opts(&path, "d".into(), PutMode::Update(v1.into()).into())
-            .await
-            .unwrap_err();
-        assert!(matches!(err, Error::Precondition { .. }), "{err}");
-
-        storage
-            .put_opts(&path, "e".into(), PutMode::Update(v2.clone().into()).into())
-            .await
-            .unwrap();
-
-        let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
-        assert_eq!(b.as_ref(), b"e");
-
-        // Update not exists
-        let path = Path::from("I don't exist");
-        let err = storage
-            .put_opts(&path, "e".into(), PutMode::Update(v2.into()).into())
-            .await
-            .unwrap_err();
-        assert!(matches!(err, Error::Precondition { .. }), "{err}");
-
-        const NUM_WORKERS: usize = 5;
-        const NUM_INCREMENTS: usize = 10;
-
-        let path = Path::from(format!("RACE-{suffix}"));
-        let mut futures: FuturesUnordered<_> = (0..NUM_WORKERS)
-            .map(|_| async {
-                for _ in 0..NUM_INCREMENTS {
-                    loop {
-                        match storage.get(&path).await {
-                            Ok(r) => {
-                                let mode = PutMode::Update(UpdateVersion {
-                                    e_tag: r.meta.e_tag.clone(),
-                                    version: r.meta.version.clone(),
-                                });
-
-                                let b = r.bytes().await.unwrap();
-                                let v: usize = std::str::from_utf8(&b).unwrap().parse().unwrap();
-                                let new = (v + 1).to_string();
-
-                                match storage.put_opts(&path, new.into(), mode.into()).await {
-                                    Ok(_) => break,
-                                    Err(Error::Precondition { .. }) => continue,
-                                    Err(e) => return Err(e),
-                                }
-                            }
-                            Err(Error::NotFound { .. }) => {
-                                let mode = PutMode::Create;
-                                match storage.put_opts(&path, "1".into(), mode.into()).await {
-                                    Ok(_) => break,
-                                    Err(Error::AlreadyExists { .. }) => continue,
-                                    Err(e) => return Err(e),
-                                }
-                            }
-                            Err(e) => return Err(e),
-                        }
-                    }
-                }
-                Ok(())
-            })
-            .collect();
-
-        while futures.next().await.transpose().unwrap().is_some() {}
-        let b = storage.get(&path).await.unwrap().bytes().await.unwrap();
-        let v = std::str::from_utf8(&b).unwrap().parse::<usize>().unwrap();
-        assert_eq!(v, NUM_WORKERS * NUM_INCREMENTS);
-    }
-
-    /// Returns a chunk of length `chunk_length`
-    fn get_chunk(chunk_length: usize) -> Bytes {
-        let mut data = vec![0_u8; chunk_length];
-        let mut rng = thread_rng();
-        // Set a random selection of bytes
-        for _ in 0..1000 {
-            data[rng.gen_range(0..chunk_length)] = rng.gen();
-        }
-        data.into()
-    }
-
-    /// Returns `num_chunks` of length `chunks`
-    fn get_chunks(chunk_length: usize, num_chunks: usize) -> Vec<Bytes> {
-        (0..num_chunks).map(|_| get_chunk(chunk_length)).collect()
-    }
-
-    pub(crate) async fn stream_get(storage: &DynObjectStore) {
-        let location = Path::from("test_dir/test_upload_file.txt");
-
-        // Can write to storage
-        let data = get_chunks(5 * 1024 * 1024, 3);
-        let bytes_expected = data.concat();
-        let mut upload = storage.put_multipart(&location).await.unwrap();
-        let uploads = data.into_iter().map(|x| upload.put_part(x.into()));
-        futures::future::try_join_all(uploads).await.unwrap();
-
-        // Object should not yet exist in store
-        let meta_res = storage.head(&location).await;
-        assert!(meta_res.is_err());
-        assert!(matches!(
-            meta_res.unwrap_err(),
-            crate::Error::NotFound { .. }
-        ));
-
-        let files = flatten_list_stream(storage, None).await.unwrap();
-        assert_eq!(&files, &[]);
-
-        let result = storage.list_with_delimiter(None).await.unwrap();
-        assert_eq!(&result.objects, &[]);
-
-        upload.complete().await.unwrap();
-
-        let bytes_written = storage.get(&location).await.unwrap().bytes().await.unwrap();
-        assert_eq!(bytes_expected, bytes_written);
-
-        // Can overwrite some storage
-        // Sizes chosen to ensure we write three parts
-        let data = get_chunks(3_200_000, 7);
-        let bytes_expected = data.concat();
-        let upload = storage.put_multipart(&location).await.unwrap();
-        let mut writer = WriteMultipart::new(upload);
-        for chunk in &data {
-            writer.write(chunk)
-        }
-        writer.finish().await.unwrap();
-        let bytes_written = storage.get(&location).await.unwrap().bytes().await.unwrap();
-        assert_eq!(bytes_expected, bytes_written);
-
-        // We can abort an empty write
-        let location = Path::from("test_dir/test_abort_upload.txt");
-        let mut upload = storage.put_multipart(&location).await.unwrap();
-        upload.abort().await.unwrap();
-        let get_res = storage.get(&location).await;
-        assert!(get_res.is_err());
-        assert!(matches!(
-            get_res.unwrap_err(),
-            crate::Error::NotFound { .. }
-        ));
-
-        // We can abort an in-progress write
-        let mut upload = storage.put_multipart(&location).await.unwrap();
-        upload
-            .put_part(data.first().unwrap().clone().into())
-            .await
-            .unwrap();
-
-        upload.abort().await.unwrap();
-        let get_res = storage.get(&location).await;
-        assert!(get_res.is_err());
-        assert!(matches!(
-            get_res.unwrap_err(),
-            crate::Error::NotFound { .. }
-        ));
-    }
-
-    pub(crate) async fn list_uses_directories_correctly(storage: &DynObjectStore) {
-        delete_fixtures(storage).await;
-
-        let content_list = flatten_list_stream(storage, None).await.unwrap();
-        assert!(
-            content_list.is_empty(),
-            "Expected list to be empty; found: {content_list:?}"
-        );
-
-        let location1 = Path::from("foo/x.json");
-        let location2 = Path::from("foo.bar/y.json");
-
-        let data = PutPayload::from("arbitrary data");
-        storage.put(&location1, data.clone()).await.unwrap();
-        storage.put(&location2, data).await.unwrap();
-
-        let prefix = Path::from("foo");
-        let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
-        assert_eq!(content_list, &[location1.clone()]);
-
-        let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
-        assert_eq!(result.objects.len(), 1);
-        assert_eq!(result.objects[0].location, location1);
-        assert_eq!(result.common_prefixes, &[]);
-
-        // Listing an existing path (file) should return an empty list:
-        // https://github.com/apache/arrow-rs/issues/3712
-        let content_list = flatten_list_stream(storage, Some(&location1))
-            .await
-            .unwrap();
-        assert_eq!(content_list, &[]);
-
-        let list = storage.list_with_delimiter(Some(&location1)).await.unwrap();
-        assert_eq!(list.objects, &[]);
-        assert_eq!(list.common_prefixes, &[]);
-
-        let prefix = Path::from("foo/x");
-        let content_list = flatten_list_stream(storage, Some(&prefix)).await.unwrap();
-        assert_eq!(content_list, &[]);
-
-        let list = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
-        assert_eq!(list.objects, &[]);
-        assert_eq!(list.common_prefixes, &[]);
-    }
-
-    pub(crate) async fn list_with_delimiter(storage: &DynObjectStore) {
-        delete_fixtures(storage).await;
-
-        // ==================== check: store is empty ====================
-        let content_list = flatten_list_stream(storage, None).await.unwrap();
-        assert!(content_list.is_empty());
-
-        // ==================== do: create files ====================
-        let data = Bytes::from("arbitrary data");
-
-        let files: Vec<_> = [
-            "test_file",
-            "mydb/wb/000/000/000.segment",
-            "mydb/wb/000/000/001.segment",
-            "mydb/wb/000/000/002.segment",
-            "mydb/wb/001/001/000.segment",
-            "mydb/wb/foo.json",
-            "mydb/wbwbwb/111/222/333.segment",
-            "mydb/data/whatevs",
-        ]
-        .iter()
-        .map(|&s| Path::from(s))
-        .collect();
-
-        for f in &files {
-            storage.put(f, data.clone().into()).await.unwrap();
-        }
-
-        // ==================== check: prefix-list `mydb/wb` (directory) ====================
-        let prefix = Path::from("mydb/wb");
-
-        let expected_000 = Path::from("mydb/wb/000");
-        let expected_001 = Path::from("mydb/wb/001");
-        let expected_location = Path::from("mydb/wb/foo.json");
-
-        let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
-
-        assert_eq!(result.common_prefixes, vec![expected_000, expected_001]);
-        assert_eq!(result.objects.len(), 1);
-
-        let object = &result.objects[0];
-
-        assert_eq!(object.location, expected_location);
-        assert_eq!(object.size, data.len());
-
-        // ==================== check: prefix-list `mydb/wb/000/000/001` (partial filename doesn't match) ====================
-        let prefix = Path::from("mydb/wb/000/000/001");
-
-        let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
-        assert!(result.common_prefixes.is_empty());
-        assert_eq!(result.objects.len(), 0);
-
-        // ==================== check: prefix-list `not_there` (non-existing prefix) ====================
-        let prefix = Path::from("not_there");
-
-        let result = storage.list_with_delimiter(Some(&prefix)).await.unwrap();
-        assert!(result.common_prefixes.is_empty());
-        assert!(result.objects.is_empty());
-
-        // ==================== do: remove all files ====================
-        for f in &files {
-            storage.delete(f).await.unwrap();
-        }
-
-        // ==================== check: store is empty ====================
-        let content_list = flatten_list_stream(storage, None).await.unwrap();
-        assert!(content_list.is_empty());
-    }
-
-    pub(crate) async fn get_nonexistent_object(
-        storage: &DynObjectStore,
-        location: Option<Path>,
-    ) -> crate::Result<Bytes> {
-        let location = location.unwrap_or_else(|| Path::from("this_file_should_not_exist"));
-
-        let err = storage.head(&location).await.unwrap_err();
-        assert!(matches!(err, crate::Error::NotFound { .. }));
-
-        storage.get(&location).await?.bytes().await
-    }
-
-    pub(crate) async fn rename_and_copy(storage: &DynObjectStore) {
-        // Create two objects
-        let path1 = Path::from("test1");
-        let path2 = Path::from("test2");
-        let contents1 = Bytes::from("cats");
-        let contents2 = Bytes::from("dogs");
-
-        // copy() make both objects identical
-        storage.put(&path1, contents1.clone().into()).await.unwrap();
-        storage.put(&path2, contents2.clone().into()).await.unwrap();
-        storage.copy(&path1, &path2).await.unwrap();
-        let new_contents = storage.get(&path2).await.unwrap().bytes().await.unwrap();
-        assert_eq!(&new_contents, &contents1);
-
-        // rename() copies contents and deletes original
-        storage.put(&path1, contents1.clone().into()).await.unwrap();
-        storage.put(&path2, contents2.clone().into()).await.unwrap();
-        storage.rename(&path1, &path2).await.unwrap();
-        let new_contents = storage.get(&path2).await.unwrap().bytes().await.unwrap();
-        assert_eq!(&new_contents, &contents1);
-        let result = storage.get(&path1).await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
-
-        // Clean up
-        storage.delete(&path2).await.unwrap();
-    }
-
-    pub(crate) async fn copy_if_not_exists(storage: &DynObjectStore) {
-        // Create two objects
-        let path1 = Path::from("test1");
-        let path2 = Path::from("not_exists_nested/test2");
-        let contents1 = Bytes::from("cats");
-        let contents2 = Bytes::from("dogs");
-
-        // copy_if_not_exists() errors if destination already exists
-        storage.put(&path1, contents1.clone().into()).await.unwrap();
-        storage.put(&path2, contents2.clone().into()).await.unwrap();
-        let result = storage.copy_if_not_exists(&path1, &path2).await;
-        assert!(result.is_err());
-        assert!(matches!(
-            result.unwrap_err(),
-            crate::Error::AlreadyExists { .. }
-        ));
-
-        // copy_if_not_exists() copies contents and allows deleting original
-        storage.delete(&path2).await.unwrap();
-        storage.copy_if_not_exists(&path1, &path2).await.unwrap();
-        storage.delete(&path1).await.unwrap();
-        let new_contents = storage.get(&path2).await.unwrap().bytes().await.unwrap();
-        assert_eq!(&new_contents, &contents1);
-        let result = storage.get(&path1).await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
-
-        // Clean up
-        storage.delete(&path2).await.unwrap();
-    }
-
-    pub(crate) async fn copy_rename_nonexistent_object(storage: &DynObjectStore) {
-        // Create empty source object
-        let path1 = Path::from("test1");
-
-        // Create destination object
-        let path2 = Path::from("test2");
-        storage.put(&path2, "hello".into()).await.unwrap();
-
-        // copy() errors if source does not exist
-        let result = storage.copy(&path1, &path2).await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
-
-        // rename() errors if source does not exist
-        let result = storage.rename(&path1, &path2).await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
-
-        // copy_if_not_exists() errors if source does not exist
-        let result = storage.copy_if_not_exists(&path1, &path2).await;
-        assert!(result.is_err());
-        assert!(matches!(result.unwrap_err(), crate::Error::NotFound { .. }));
-
-        // Clean up
-        storage.delete(&path2).await.unwrap();
-    }
-
-    pub(crate) async fn multipart(storage: &dyn ObjectStore, multipart: &dyn MultipartStore) {
-        let path = Path::from("test_multipart");
-        let chunk_size = 5 * 1024 * 1024;
-
-        let chunks = get_chunks(chunk_size, 2);
-
-        let id = multipart.create_multipart(&path).await.unwrap();
-
-        let parts: Vec<_> = futures::stream::iter(chunks)
-            .enumerate()
-            .map(|(idx, b)| multipart.put_part(&path, &id, idx, b.into()))
-            .buffered(2)
-            .try_collect()
-            .await
-            .unwrap();
-
-        multipart
-            .complete_multipart(&path, &id, parts)
-            .await
-            .unwrap();
-
-        let meta = storage.head(&path).await.unwrap();
-        assert_eq!(meta.size, chunk_size * 2);
-
-        // Empty case
-        let path = Path::from("test_empty_multipart");
-
-        let id = multipart.create_multipart(&path).await.unwrap();
-
-        let parts = vec![];
-
-        multipart
-            .complete_multipart(&path, &id, parts)
-            .await
-            .unwrap();
-
-        let meta = storage.head(&path).await.unwrap();
-        assert_eq!(meta.size, 0);
+    /// Test that the returned stream does not borrow the lifetime of Path
+    fn list_store<'a>(
+        store: &'a dyn ObjectStore,
+        path_str: &str,
+    ) -> BoxStream<'a, Result<ObjectMeta>> {
+        let path = Path::from(path_str);
+        store.list(Some(&path))
     }
 
     #[cfg(any(feature = "azure", feature = "aws"))]
-    pub(crate) async fn signing<T>(integration: &T)
+    pub async fn signing<T>(integration: &T)
     where
-        T: ObjectStore + crate::signer::Signer,
+        T: ObjectStore + signer::Signer,
     {
         use reqwest::Method;
         use std::time::Duration;
@@ -2367,7 +1337,7 @@ mod tests {
     }
 
     #[cfg(any(feature = "aws", feature = "azure"))]
-    pub(crate) async fn tagging<F, Fut>(storage: Arc<dyn ObjectStore>, validate: bool, get_tags: F)
+    pub async fn tagging<F, Fut>(storage: Arc<dyn ObjectStore>, validate: bool, get_tags: F)
     where
         F: Fn(Path) -> Fut + Send + Sync,
         Fut: std::future::Future<Output = Result<reqwest::Response>> + Send,
@@ -2444,24 +1414,6 @@ mod tests {
         }
     }
 
-    async fn delete_fixtures(storage: &DynObjectStore) {
-        let paths = storage.list(None).map_ok(|meta| meta.location).boxed();
-        storage
-            .delete_stream(paths)
-            .try_collect::<Vec<_>>()
-            .await
-            .unwrap();
-    }
-
-    /// Test that the returned stream does not borrow the lifetime of Path
-    fn list_store<'a>(
-        store: &'a dyn ObjectStore,
-        path_str: &str,
-    ) -> BoxStream<'a, Result<ObjectMeta>> {
-        let path = Path::from(path_str);
-        store.list(Some(&path))
-    }
-
     #[tokio::test]
     async fn test_list_lifetimes() {
         let store = memory::InMemory::new();
diff --git a/object_store/src/limit.rs b/object_store/src/limit.rs
index f3e1d4296..64b96ad1a 100644
--- a/object_store/src/limit.rs
+++ b/object_store/src/limit.rs
@@ -273,9 +273,9 @@ impl MultipartUpload for LimitUpload {
 
 #[cfg(test)]
 mod tests {
+    use crate::integration::*;
     use crate::limit::LimitStore;
     use crate::memory::InMemory;
-    use crate::tests::*;
     use crate::ObjectStore;
     use futures::stream::StreamExt;
     use std::pin::Pin;
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 8dec5bee0..95b50d674 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -1005,8 +1005,7 @@ mod tests {
     use futures::TryStreamExt;
     use tempfile::{NamedTempFile, TempDir};
 
-    use crate::test_util::flatten_list_stream;
-    use crate::tests::*;
+    use crate::integration::*;
 
     use super::*;
 
diff --git a/object_store/src/memory.rs b/object_store/src/memory.rs
index daf14e175..0d72983b0 100644
--- a/object_store/src/memory.rs
+++ b/object_store/src/memory.rs
@@ -528,7 +528,7 @@ impl MultipartUpload for InMemoryUpload {
 
 #[cfg(test)]
 mod tests {
-    use crate::tests::*;
+    use crate::integration::*;
 
     use super::*;
 
diff --git a/object_store/src/parse.rs b/object_store/src/parse.rs
index 5549fd3a3..e5d5149fb 100644
--- a/object_store/src/parse.rs
+++ b/object_store/src/parse.rs
@@ -25,15 +25,9 @@ use url::Url;
 
 #[derive(Debug, Snafu)]
 enum Error {
-    #[snafu(display("Unable to convert URL \"{}\" to filesystem path", url))]
-    InvalidUrl { url: Url },
-
     #[snafu(display("Unable to recognise URL \"{}\"", url))]
     Unrecognised { url: Url },
 
-    #[snafu(display("Feature {scheme:?} not enabled"))]
-    NotEnabled { scheme: ObjectStoreScheme },
-
     #[snafu(context(false))]
     Path { source: crate::path::Error },
 }
diff --git a/object_store/src/prefix.rs b/object_store/src/prefix.rs
index 7c9ea5804..9b10fea5e 100644
--- a/object_store/src/prefix.rs
+++ b/object_store/src/prefix.rs
@@ -204,9 +204,8 @@ impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::integration::*;
     use crate::local::LocalFileSystem;
-    use crate::test_util::flatten_list_stream;
-    use crate::tests::*;
 
     use tempfile::TempDir;
 
diff --git a/object_store/src/throttle.rs b/object_store/src/throttle.rs
index 38b6d7c3b..d07276c3d 100644
--- a/object_store/src/throttle.rs
+++ b/object_store/src/throttle.rs
@@ -398,7 +398,7 @@ impl MultipartUpload for ThrottledUpload {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{memory::InMemory, tests::*, GetResultPayload};
+    use crate::{integration::*, memory::InMemory, GetResultPayload};
     use futures::TryStreamExt;
     use tokio::time::Duration;
     use tokio::time::Instant;
diff --git a/parquet/src/bloom_filter/mod.rs b/parquet/src/bloom_filter/mod.rs
index d99c72519..f746d8b9a 100644
--- a/parquet/src/bloom_filter/mod.rs
+++ b/parquet/src/bloom_filter/mod.rs
@@ -16,7 +16,61 @@
 // under the License.
 
 //! Bloom filter implementation specific to Parquet, as described
-//! in the [spec](https://github.com/apache/parquet-format/blob/master/BloomFilter.md).
+//! in the [spec][parquet-bf-spec].
+//!
+//! # Bloom Filter Size
+//!
+//! Parquet uses the [Split Block Bloom Filter][sbbf-paper] (SBBF) as its bloom filter
+//! implementation. For each column upon which bloom filters are enabled, the offset and length of an SBBF
+//! is stored in  the metadata for each row group in the parquet file. The size of each filter is
+//! initialized using a calculation based on the desired number of distinct values (NDV) and false
+//! positive probability (FPP). The FPP for a SBBF can be approximated as<sup>[1][bf-formulae]</sup>:
+//!
+//! ```text
+//! f = (1 - e^(-k * n / m))^k
+//! ```
+//!
+//! Where, `f` is the FPP, `k` the number of hash functions, `n` the NDV, and `m` the total number
+//! of bits in the bloom filter. This can be re-arranged to determine the total number of bits
+//! required to achieve a given FPP and NDV:
+//!
+//! ```text
+//! m = -k * n / ln(1 - f^(1/k))
+//! ```
+//!
+//! SBBFs use eight hash functions to cleanly fit in SIMD lanes<sup>[2][sbbf-paper]</sup>, therefore
+//! `k` is set to 8. The SBBF will spread those `m` bits accross a set of `b` blocks that
+//! are each 256 bits, i.e., 32 bytes, in size. The number of blocks is chosen as:
+//!
+//! ```text
+//! b = NP2(m/8) / 32
+//! ```
+//!
+//! Where, `NP2` denotes *the next power of two*, and `m` is divided by 8 to be represented as bytes.
+//!
+//! Here is a table of calculated sizes for various FPP and NDV:
+//!
+//! | NDV       | FPP       | b       | Size (KB) |
+//! |-----------|-----------|---------|-----------|
+//! | 10,000    | 0.1       | 256     | 8         |
+//! | 10,000    | 0.01      | 512     | 16        |
+//! | 10,000    | 0.001     | 1,024   | 32        |
+//! | 10,000    | 0.0001    | 1,024   | 32        |
+//! | 100,000   | 0.1       | 4,096   | 128       |
+//! | 100,000   | 0.01      | 4,096   | 128       |
+//! | 100,000   | 0.001     | 8,192   | 256       |
+//! | 100,000   | 0.0001    | 16,384  | 512       |
+//! | 100,000   | 0.00001   | 16,384  | 512       |
+//! | 1,000,000 | 0.1       | 32,768  | 1,024     |
+//! | 1,000,000 | 0.01      | 65,536  | 2,048     |
+//! | 1,000,000 | 0.001     | 65,536  | 2,048     |
+//! | 1,000,000 | 0.0001    | 131,072 | 4,096     |
+//! | 1,000,000 | 0.00001   | 131,072 | 4,096     |
+//! | 1,000,000 | 0.000001  | 262,144 | 8,192     |
+//!
+//! [parquet-bf-spec]: https://github.com/apache/parquet-format/blob/master/BloomFilter.md
+//! [sbbf-paper]: https://arxiv.org/pdf/2101.01719
+//! [bf-formulae]: http://tfk.mit.edu/pdf/bloom.pdf
 
 use crate::data_type::AsBytes;
 use crate::errors::ParquetError;
@@ -179,7 +233,7 @@ fn num_of_bits_from_ndv_fpp(ndv: u64, fpp: f64) -> usize {
 
 impl Sbbf {
     /// Create a new [Sbbf] with given number of distinct values and false positive probability.
-    /// Will panic if `fpp` is greater than 1.0 or less than 0.0.
+    /// Will return an error if `fpp` is greater than or equal to 1.0 or less than 0.0.
     pub(crate) fn new_with_ndv_fpp(ndv: u64, fpp: f64) -> Result<Self, ParquetError> {
         if !(0.0..1.0).contains(&fpp) {
             return Err(ParquetError::General(format!(
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 9ef505cc7..e084d9ba4 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -1172,13 +1172,6 @@ fn compare_greater<T: ParquetValueType>(descr: &ColumnDescriptor, a: &T, b: &T)
 // https://github.com/apache/parquet-mr/blob/master/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java
 // https://github.com/apache/parquet-mr/blob/master/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV2ValuesWriterFactory.java
 
-/// Trait to define default encoding for types, including whether or not the type
-/// supports dictionary encoding.
-trait EncodingWriteSupport {
-    /// Returns true if dictionary is supported for column writer, false otherwise.
-    fn has_dictionary_support(props: &WriterProperties) -> bool;
-}
-
 /// Returns encoding for a column when no other encoding is provided in writer properties.
 fn fallback_encoding(kind: Type, props: &WriterProperties) -> Encoding {
     match (kind, props.writer_version()) {
diff --git a/parquet/src/encodings/encoding/mod.rs b/parquet/src/encodings/encoding/mod.rs
index 9099187a5..d797b3cb2 100644
--- a/parquet/src/encodings/encoding/mod.rs
+++ b/parquet/src/encodings/encoding/mod.rs
@@ -24,7 +24,7 @@ use crate::data_type::private::ParquetValueType;
 use crate::data_type::*;
 use crate::encodings::rle::RleEncoder;
 use crate::errors::{ParquetError, Result};
-use crate::util::bit_util::{self, num_required_bits, BitWriter};
+use crate::util::bit_util::{num_required_bits, BitWriter};
 
 use bytes::Bytes;
 pub use dict_encoder::DictEncoder;
@@ -47,12 +47,13 @@ pub trait Encoder<T: DataType>: Send {
     /// identified by `valid_bits`.
     ///
     /// Returns the number of non-null values encoded.
+    #[cfg(test)]
     fn put_spaced(&mut self, values: &[T::T], valid_bits: &[u8]) -> Result<usize> {
         let num_values = values.len();
         let mut buffer = Vec::with_capacity(num_values);
         // TODO: this is pretty inefficient. Revisit in future.
         for (i, item) in values.iter().enumerate().take(num_values) {
-            if bit_util::get_bit(valid_bits, i) {
+            if crate::util::bit_util::get_bit(valid_bits, i) {
                 buffer.push(item.clone());
             }
         }
@@ -727,6 +728,7 @@ mod tests {
     use crate::encodings::decoding::{get_decoder, Decoder, DictDecoder, PlainDecoder};
     use crate::schema::types::{ColumnDescPtr, ColumnDescriptor, ColumnPath, Type as SchemaType};
     use crate::util::test_common::rand_gen::{random_bytes, RandGen};
+    use crate::util::bit_util;
 
     const TEST_SET_SIZE: usize = 1024;
 
diff --git a/parquet/src/file/statistics.rs b/parquet/src/file/statistics.rs
index 1bc003d48..d24b91741 100644
--- a/parquet/src/file/statistics.rs
+++ b/parquet/src/file/statistics.rs
@@ -271,8 +271,8 @@ pub fn to_thrift(stats: Option<&Statistics>) -> Option<TStatistics> {
 
     if stats.is_min_max_backwards_compatible() {
         // Copy to deprecated min, max values for compatibility with older readers
-        thrift_stats.min = min.clone();
-        thrift_stats.max = max.clone();
+        thrift_stats.min.clone_from(&min);
+        thrift_stats.max.clone_from(&max);
     }
 
     if !stats.is_min_max_deprecated() {


Test patch:
diff --git a/arrow-flight/tests/client.rs b/arrow-flight/tests/client.rs
index 47565334c..478938d93 100644
--- a/arrow-flight/tests/client.rs
+++ b/arrow-flight/tests/client.rs
@@ -493,7 +493,7 @@ async fn test_do_exchange() {
             .set_do_exchange_response(output_flight_data.clone().into_iter().map(Ok).collect());
 
         let response_stream = client
-            .do_exchange(futures::stream::iter(input_flight_data.clone()))
+            .do_exchange(futures::stream::iter(input_flight_data.clone()).map(Ok))
             .await
             .expect("error making request");
 
@@ -528,7 +528,7 @@ async fn test_do_exchange_error() {
         let input_flight_data = test_flight_data().await;
 
         let response = client
-            .do_exchange(futures::stream::iter(input_flight_data.clone()))
+            .do_exchange(futures::stream::iter(input_flight_data.clone()).map(Ok))
             .await;
         let response = match response {
             Ok(_) => panic!("unexpected success"),
@@ -572,7 +572,7 @@ async fn test_do_exchange_error_stream() {
         test_server.set_do_exchange_response(response);
 
         let response_stream = client
-            .do_exchange(futures::stream::iter(input_flight_data.clone()))
+            .do_exchange(futures::stream::iter(input_flight_data.clone()).map(Ok))
             .await
             .expect("error making request");
 
@@ -593,6 +593,97 @@ async fn test_do_exchange_error_stream() {
     .await;
 }
 
+#[tokio::test]
+async fn test_do_exchange_error_stream_client() {
+    do_test(|test_server, mut client| async move {
+        client.add_header("foo-header", "bar-header-value").unwrap();
+
+        let e = Status::invalid_argument("bad arg: client");
+
+        // input stream to client sends good FlightData followed by an error
+        let input_flight_data = test_flight_data().await;
+        let input_stream = futures::stream::iter(input_flight_data.clone())
+            .map(Ok)
+            .chain(futures::stream::iter(vec![Err(FlightError::from(
+                e.clone(),
+            ))]));
+
+        let output_flight_data = FlightData::new()
+            .with_descriptor(FlightDescriptor::new_cmd("Sample command"))
+            .with_data_body("body".as_bytes())
+            .with_data_header("header".as_bytes())
+            .with_app_metadata("metadata".as_bytes());
+
+        // server responds with one good message
+        let response = vec![Ok(output_flight_data)];
+        test_server.set_do_exchange_response(response);
+
+        let response_stream = client
+            .do_exchange(input_stream)
+            .await
+            .expect("error making request");
+
+        let response: Result<Vec<_>, _> = response_stream.try_collect().await;
+        let response = match response {
+            Ok(_) => panic!("unexpected success"),
+            Err(e) => e,
+        };
+
+        // expect to the error made from the client
+        expect_status(response, e);
+        // server still got the request messages until the client sent the error
+        assert_eq!(
+            test_server.take_do_exchange_request(),
+            Some(input_flight_data)
+        );
+        ensure_metadata(&client, &test_server);
+    })
+    .await;
+}
+
+#[tokio::test]
+async fn test_do_exchange_error_client_and_server() {
+    do_test(|test_server, mut client| async move {
+        client.add_header("foo-header", "bar-header-value").unwrap();
+
+        let e_client = Status::invalid_argument("bad arg: client");
+        let e_server = Status::invalid_argument("bad arg: server");
+
+        // input stream to client sends good FlightData followed by an error
+        let input_flight_data = test_flight_data().await;
+        let input_stream = futures::stream::iter(input_flight_data.clone())
+            .map(Ok)
+            .chain(futures::stream::iter(vec![Err(FlightError::from(
+                e_client.clone(),
+            ))]));
+
+        // server responds with an error (e.g. because it got truncated data)
+        let response = vec![Err(e_server)];
+        test_server.set_do_exchange_response(response);
+
+        let response_stream = client
+            .do_exchange(input_stream)
+            .await
+            .expect("error making request");
+
+        let response: Result<Vec<_>, _> = response_stream.try_collect().await;
+        let response = match response {
+            Ok(_) => panic!("unexpected success"),
+            Err(e) => e,
+        };
+
+        // expect to the error made from the client (not the server)
+        expect_status(response, e_client);
+        // server still got the request messages until the client sent the error
+        assert_eq!(
+            test_server.take_do_exchange_request(),
+            Some(input_flight_data)
+        );
+        ensure_metadata(&client, &test_server);
+    })
+    .await;
+}
+
 #[tokio::test]
 async fn test_get_schema() {
     do_test(|test_server, mut client| async move {
diff --git a/arrow-integration-testing/src/bin/arrow-json-integration-test.rs b/arrow-integration-testing/src/bin/arrow-json-integration-test.rs
index 9f1abb16a..cc3dd2110 100644
--- a/arrow-integration-testing/src/bin/arrow-json-integration-test.rs
+++ b/arrow-integration-testing/src/bin/arrow-json-integration-test.rs
@@ -40,7 +40,13 @@ struct Args {
     arrow: String,
     #[clap(short, long, help("Path to JSON file"))]
     json: String,
-    #[clap(value_enum, short, long, default_value_t = Mode::Validate, help="Mode of integration testing tool")]
+    #[clap(
+        value_enum,
+        short,
+        long,
+        default_value = "VALIDATE",
+        help = "Mode of integration testing tool"
+    )]
     mode: Mode,
     #[clap(short, long)]
     verbose: bool,