Repository: apache/arrow-rs
Pull request #5247 (2023-12-26 22:57:38)

DynamoDB ConditionalPut
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes #.

# Rationale for this change
 
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->

Builds on #4918 and extends it to conditional put.

# What changes are included in this PR?

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?


<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->

<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->


tustvold: @roeap when you get a chance, perhaps you could give this a look over

Patch:
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index 1604a7be4..23c4f4af2 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -63,7 +63,8 @@ jobs:
       ARROW_INTEGRATION_CSHARP: ON
       ARROW_INTEGRATION_GO: ON
       ARROW_INTEGRATION_JAVA: ON
-      ARROW_INTEGRATION_JS: ON
+      # https://github.com/apache/arrow/issues/39395
+      ARROW_INTEGRATION_JS: OFF
       # https://github.com/apache/arrow/pull/38403/files#r1371281630
       ARCHERY_INTEGRATION_WITH_RUST: ON
       # These are necessary because the github runner overrides $HOME
diff --git a/.github/workflows/object_store.yml b/.github/workflows/object_store.yml
index 313d15809..0257d86d9 100644
--- a/.github/workflows/object_store.yml
+++ b/.github/workflows/object_store.yml
@@ -113,6 +113,7 @@ jobs:
       AWS_ENDPOINT: http://localhost:4566
       AWS_ALLOW_HTTP: true
       AWS_COPY_IF_NOT_EXISTS: dynamo:test-table:2000
+      AWS_CONDITIONAL_PUT: dynamo:test-table:2000
       HTTP_URL: "http://localhost:8080"
       GOOGLE_BUCKET: test-bucket
       GOOGLE_SERVICE_ACCOUNT: "/tmp/gcs.json"
@@ -137,7 +138,7 @@ jobs:
           docker run -d -p 4566:4566 localstack/localstack:3.0.1
           docker run -d -p 1338:1338 amazon/amazon-ec2-metadata-mock:v1.9.2 --imdsv2
           aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket
-          aws --endpoint-url=http://localhost:4566 dynamodb create-table --table-name test-table --key-schema AttributeName=key,KeyType=HASH --attribute-definitions AttributeName=key,AttributeType=S --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5
+          aws --endpoint-url=http://localhost:4566 dynamodb create-table --table-name test-table --key-schema AttributeName=path,KeyType=HASH AttributeName=etag,KeyType=RANGE --attribute-definitions AttributeName=path,AttributeType=S AttributeName=etag,AttributeType=S --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5
 
       - name: Configure Azurite (Azure emulation)
         # the magical connection string is from
diff --git a/.gitignore b/.gitignore
index 52ad19cb0..0788daea0 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,6 +5,7 @@ rusty-tags.vi
 .flatbuffers/
 .idea/
 .vscode
+.devcontainer
 venv/*
 # created by doctests
 parquet/data.parquet
@@ -14,6 +15,7 @@ parquet/data.parquet
 justfile
 .prettierignore
 .env
+.editorconfig
 # local azurite file
 __azurite*
 __blobstorage__
diff --git a/arrow-array/src/array/primitive_array.rs b/arrow-array/src/array/primitive_array.rs
index 2296cebd4..a6d5c2277 100644
--- a/arrow-array/src/array/primitive_array.rs
+++ b/arrow-array/src/array/primitive_array.rs
@@ -713,9 +713,8 @@ impl<T: ArrowPrimitiveType> PrimitiveArray<T> {
     /// the semantic values of the array, e.g. 100 milliseconds in a [`TimestampNanosecondArray`]
     /// will become 100 seconds in a [`TimestampSecondArray`].
     ///
-    /// For casts that preserve the semantic value, check out the [compute kernels]
-    ///
-    /// [compute kernels](https://docs.rs/arrow/latest/arrow/compute/kernels/cast/index.html)
+    /// For casts that preserve the semantic value, check out the
+    /// [compute kernels](https://docs.rs/arrow/latest/arrow/compute/kernels/cast/index.html).
     ///
     /// ```
     /// # use arrow_array::{Int64Array, TimestampNanosecondArray};
diff --git a/arrow-data/src/equal/run.rs b/arrow-data/src/equal/run.rs
index ede172c99..6c9393ecd 100644
--- a/arrow-data/src/equal/run.rs
+++ b/arrow-data/src/equal/run.rs
@@ -43,11 +43,13 @@ pub(super) fn run_equal(
         return false;
     }
 
-    let lhs_run_ends_array = lhs.child_data().get(0).unwrap();
-    let lhs_values_array = lhs.child_data().get(1).unwrap();
+    let lhs_child_data = lhs.child_data();
+    let lhs_run_ends_array = &lhs_child_data[0];
+    let lhs_values_array = &lhs_child_data[1];
 
-    let rhs_run_ends_array = rhs.child_data().get(0).unwrap();
-    let rhs_values_array = rhs.child_data().get(1).unwrap();
+    let rhs_child_data = rhs.child_data();
+    let rhs_run_ends_array = &rhs_child_data[0];
+    let rhs_values_array = &rhs_child_data[1];
 
     if lhs_run_ends_array.len() != rhs_run_ends_array.len() {
         return false;
diff --git a/arrow-data/src/equal/utils.rs b/arrow-data/src/equal/utils.rs
index cc8194375..f1f4be447 100644
--- a/arrow-data/src/equal/utils.rs
+++ b/arrow-data/src/equal/utils.rs
@@ -67,10 +67,10 @@ pub(super) fn base_equal(lhs: &ArrayData, rhs: &ArrayData) -> bool {
                 (DataType::Struct(l_fields), DataType::Struct(r_fields))
                     if l_fields.len() == 2 && r_fields.len() == 2 =>
                 {
-                    let l_key_field = l_fields.get(0).unwrap();
-                    let r_key_field = r_fields.get(0).unwrap();
-                    let l_value_field = l_fields.get(1).unwrap();
-                    let r_value_field = r_fields.get(1).unwrap();
+                    let l_key_field = &l_fields[0];
+                    let r_key_field = &r_fields[0];
+                    let l_value_field = &l_fields[1];
+                    let r_value_field = &r_fields[1];
 
                     // We don't enforce the equality of field names
                     let data_type_equal = l_key_field.data_type() == r_key_field.data_type()
diff --git a/arrow-flight/gen/Cargo.toml b/arrow-flight/gen/Cargo.toml
index c96db2187..0cde7fa9c 100644
--- a/arrow-flight/gen/Cargo.toml
+++ b/arrow-flight/gen/Cargo.toml
@@ -32,6 +32,6 @@ publish = false
 [dependencies]
 # Pin specific version of the tonic-build dependencies to avoid auto-generated
 # (and checked in) arrow.flight.protocol.rs from changing
-proc-macro2 = { version = "=1.0.71", default-features = false }
+proc-macro2 = { version = "=1.0.75", default-features = false }
 prost-build = { version = "=0.12.3", default-features = false }
 tonic-build = { version = "=0.10.2", default-features = false, features = ["transport", "prost"] }
diff --git a/arrow-flight/src/utils.rs b/arrow-flight/src/utils.rs
index b75d61d20..32716a52e 100644
--- a/arrow-flight/src/utils.rs
+++ b/arrow-flight/src/utils.rs
@@ -53,7 +53,7 @@ pub fn flight_data_from_arrow_batch(
 
 /// Convert a slice of wire protocol `FlightData`s into a vector of `RecordBatch`es
 pub fn flight_data_to_batches(flight_data: &[FlightData]) -> Result<Vec<RecordBatch>, ArrowError> {
-    let schema = flight_data.get(0).ok_or_else(|| {
+    let schema = flight_data.first().ok_or_else(|| {
         ArrowError::CastError("Need at least one FlightData for schema".to_string())
     })?;
     let message = root_as_message(&schema.data_header[..])
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index 39365b782..8ac3a387d 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -451,7 +451,7 @@ pub fn read_dictionary(
     batch: crate::DictionaryBatch,
     schema: &Schema,
     dictionaries_by_id: &mut HashMap<i64, ArrayRef>,
-    metadata: &crate::MetadataVersion,
+    metadata: &MetadataVersion,
 ) -> Result<(), ArrowError> {
     if batch.isDelta() {
         return Err(ArrowError::InvalidArgumentError(
@@ -522,6 +522,174 @@ fn parse_message(buf: &[u8]) -> Result<Message, ArrowError> {
         .map_err(|err| ArrowError::ParseError(format!("Unable to get root as message: {err:?}")))
 }
 
+/// Read the footer length from the last 10 bytes of an Arrow IPC file
+///
+/// Expects a 4 byte footer length followed by `b"ARROW1"`
+pub fn read_footer_length(buf: [u8; 10]) -> Result<usize, ArrowError> {
+    if buf[4..] != super::ARROW_MAGIC {
+        return Err(ArrowError::ParseError(
+            "Arrow file does not contain correct footer".to_string(),
+        ));
+    }
+
+    // read footer length
+    let footer_len = i32::from_le_bytes(buf[..4].try_into().unwrap());
+    footer_len
+        .try_into()
+        .map_err(|_| ArrowError::ParseError(format!("Invalid footer length: {footer_len}")))
+}
+
+/// A low-level, push-based interface for reading an IPC file
+///
+/// For a higher-level interface see [`FileReader`]
+///
+/// ```
+/// # use std::sync::Arc;
+/// # use arrow_array::*;
+/// # use arrow_array::types::Int32Type;
+/// # use arrow_buffer::Buffer;
+/// # use arrow_ipc::convert::fb_to_schema;
+/// # use arrow_ipc::reader::{FileDecoder, read_footer_length};
+/// # use arrow_ipc::root_as_footer;
+/// # use arrow_ipc::writer::FileWriter;
+/// // Write an IPC file
+///
+/// let batch = RecordBatch::try_from_iter([
+///     ("a", Arc::new(Int32Array::from(vec![1, 2, 3])) as _),
+///     ("b", Arc::new(Int32Array::from(vec![1, 2, 3])) as _),
+///     ("c", Arc::new(DictionaryArray::<Int32Type>::from_iter(["hello", "hello", "world"])) as _),
+/// ]).unwrap();
+///
+/// let schema = batch.schema();
+///
+/// let mut out = Vec::with_capacity(1024);
+/// let mut writer = FileWriter::try_new(&mut out, schema.as_ref()).unwrap();
+/// writer.write(&batch).unwrap();
+/// writer.finish().unwrap();
+///
+/// drop(writer);
+///
+/// // Read IPC file
+///
+/// let buffer = Buffer::from_vec(out);
+/// let trailer_start = buffer.len() - 10;
+/// let footer_len = read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
+/// let footer = root_as_footer(&buffer[trailer_start - footer_len..trailer_start]).unwrap();
+///
+/// let back = fb_to_schema(footer.schema().unwrap());
+/// assert_eq!(&back, schema.as_ref());
+///
+/// let mut decoder = FileDecoder::new(schema, footer.version());
+///
+/// // Read dictionaries
+/// for block in footer.dictionaries().iter().flatten() {
+///     let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+///     let data = buffer.slice_with_length(block.offset() as _, block_len);
+///     decoder.read_dictionary(&block, &data).unwrap();
+/// }
+///
+/// // Read record batch
+/// let batches = footer.recordBatches().unwrap();
+/// assert_eq!(batches.len(), 1); // Only wrote a single batch
+///
+/// let block = batches.get(0);
+/// let block_len = block.bodyLength() as usize + block.metaDataLength() as usize;
+/// let data = buffer.slice_with_length(block.offset() as _, block_len);
+/// let back = decoder.read_record_batch(block, &data).unwrap().unwrap();
+///
+/// assert_eq!(batch, back);
+/// ```
+#[derive(Debug)]
+pub struct FileDecoder {
+    schema: SchemaRef,
+    dictionaries: HashMap<i64, ArrayRef>,
+    version: MetadataVersion,
+    projection: Option<Vec<usize>>,
+}
+
+impl FileDecoder {
+    /// Create a new [`FileDecoder`] with the given schema and version
+    pub fn new(schema: SchemaRef, version: MetadataVersion) -> Self {
+        Self {
+            schema,
+            version,
+            dictionaries: Default::default(),
+            projection: None,
+        }
+    }
+
+    /// Specify a projection
+    pub fn with_projection(mut self, projection: Vec<usize>) -> Self {
+        self.projection = Some(projection);
+        self
+    }
+
+    fn read_message<'a>(&self, buf: &'a [u8]) -> Result<Message<'a>, ArrowError> {
+        let message = parse_message(buf)?;
+
+        // some old test data's footer metadata is not set, so we account for that
+        if self.version != MetadataVersion::V1 && message.version() != self.version {
+            return Err(ArrowError::IpcError(
+                "Could not read IPC message as metadata versions mismatch".to_string(),
+            ));
+        }
+        Ok(message)
+    }
+
+    /// Read the dictionary with the given block and data buffer
+    pub fn read_dictionary(&mut self, block: &Block, buf: &Buffer) -> Result<(), ArrowError> {
+        let message = self.read_message(buf)?;
+        match message.header_type() {
+            crate::MessageHeader::DictionaryBatch => {
+                let batch = message.header_as_dictionary_batch().unwrap();
+                read_dictionary(
+                    &buf.slice(block.metaDataLength() as _),
+                    batch,
+                    &self.schema,
+                    &mut self.dictionaries,
+                    &message.version(),
+                )
+            }
+            t => Err(ArrowError::ParseError(format!(
+                "Expecting DictionaryBatch in dictionary blocks, found {t:?}."
+            ))),
+        }
+    }
+
+    /// Read the RecordBatch with the given block and data buffer
+    pub fn read_record_batch(
+        &self,
+        block: &Block,
+        buf: &Buffer,
+    ) -> Result<Option<RecordBatch>, ArrowError> {
+        let message = self.read_message(buf)?;
+        match message.header_type() {
+            crate::MessageHeader::Schema => Err(ArrowError::IpcError(
+                "Not expecting a schema when messages are read".to_string(),
+            )),
+            crate::MessageHeader::RecordBatch => {
+                let batch = message.header_as_record_batch().ok_or_else(|| {
+                    ArrowError::IpcError("Unable to read IPC message as record batch".to_string())
+                })?;
+                // read the block that makes up the record batch into a buffer
+                read_record_batch(
+                    &buf.slice(block.metaDataLength() as _),
+                    batch,
+                    self.schema.clone(),
+                    &self.dictionaries,
+                    self.projection.as_deref(),
+                    &message.version(),
+                )
+                .map(Some)
+            }
+            crate::MessageHeader::NONE => Ok(None),
+            t => Err(ArrowError::InvalidArgumentError(format!(
+                "Reading types other than record batches not yet supported, unable to read {t:?}"
+            ))),
+        }
+    }
+}
+
 /// Build an Arrow [`FileReader`] with custom options.
 #[derive(Debug)]
 pub struct FileReaderBuilder {
@@ -599,17 +767,10 @@ impl FileReaderBuilder {
         reader.seek(SeekFrom::End(-10))?;
         reader.read_exact(&mut buffer)?;
 
-        if buffer[4..] != super::ARROW_MAGIC {
-            return Err(ArrowError::ParseError(
-                "Arrow file does not contain correct footer".to_string(),
-            ));
-        }
-
-        // read footer length
-        let footer_len = i32::from_le_bytes(buffer[..4].try_into().unwrap());
+        let footer_len = read_footer_length(buffer)?;
 
         // read footer
-        let mut footer_data = vec![0; footer_len as usize];
+        let mut footer_data = vec![0; footer_len];
         reader.seek(SeekFrom::End(-10 - footer_len as i64))?;
         reader.read_exact(&mut footer_data)?;
 
@@ -641,50 +802,26 @@ impl FileReaderBuilder {
             }
         }
 
+        let mut decoder = FileDecoder::new(Arc::new(schema), footer.version());
+        if let Some(projection) = self.projection {
+            decoder = decoder.with_projection(projection)
+        }
+
         // Create an array of optional dictionary value arrays, one per field.
-        let mut dictionaries_by_id = HashMap::new();
         if let Some(dictionaries) = footer.dictionaries() {
             for block in dictionaries {
                 let buf = read_block(&mut reader, block)?;
-                let message = parse_message(&buf)?;
-
-                match message.header_type() {
-                    crate::MessageHeader::DictionaryBatch => {
-                        let batch = message.header_as_dictionary_batch().unwrap();
-                        read_dictionary(
-                            &buf.slice(block.metaDataLength() as _),
-                            batch,
-                            &schema,
-                            &mut dictionaries_by_id,
-                            &message.version(),
-                        )?;
-                    }
-                    t => {
-                        return Err(ArrowError::ParseError(format!(
-                            "Expecting DictionaryBatch in dictionary blocks, found {t:?}."
-                        )));
-                    }
-                }
+                decoder.read_dictionary(block, &buf)?;
             }
         }
-        let projection = match self.projection {
-            Some(projection_indices) => {
-                let schema = schema.project(&projection_indices)?;
-                Some((projection_indices, schema))
-            }
-            _ => None,
-        };
 
         Ok(FileReader {
             reader,
-            schema: Arc::new(schema),
             blocks: blocks.iter().copied().collect(),
             current_block: 0,
             total_blocks,
-            dictionaries_by_id,
-            metadata_version: footer.version(),
+            decoder,
             custom_metadata,
-            projection,
         })
     }
 }
@@ -694,13 +831,13 @@ pub struct FileReader<R: Read + Seek> {
     /// Buffered file reader that supports reading and seeking
     reader: R,
 
-    /// The schema that is read from the file header
-    schema: SchemaRef,
+    /// The decoder
+    decoder: FileDecoder,
 
     /// The blocks in the file
     ///
     /// A block indicates the regions in the file to read to get data
-    blocks: Vec<crate::Block>,
+    blocks: Vec<Block>,
 
     /// A counter to keep track of the current block that should be read
     current_block: usize,
@@ -708,31 +845,17 @@ pub struct FileReader<R: Read + Seek> {
     /// The total number of blocks, which may contain record batches and other types
     total_blocks: usize,
 
-    /// Optional dictionaries for each schema field.
-    ///
-    /// Dictionaries may be appended to in the streaming format.
-    dictionaries_by_id: HashMap<i64, ArrayRef>,
-
-    /// Metadata version
-    metadata_version: crate::MetadataVersion,
-
     /// User defined metadata
     custom_metadata: HashMap<String, String>,
-
-    /// Optional projection and projected_schema
-    projection: Option<(Vec<usize>, Schema)>,
 }
 
 impl<R: Read + Seek> fmt::Debug for FileReader<R> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> std::result::Result<(), fmt::Error> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> Result<(), fmt::Error> {
         f.debug_struct("FileReader<R>")
-            .field("schema", &self.schema)
+            .field("decoder", &self.decoder)
             .field("blocks", &self.blocks)
             .field("current_block", &self.current_block)
             .field("total_blocks", &self.total_blocks)
-            .field("dictionaries_by_id", &self.dictionaries_by_id)
-            .field("metadata_version", &self.metadata_version)
-            .field("projection", &self.projection)
             .finish_non_exhaustive()
     }
 }
@@ -761,7 +884,7 @@ impl<R: Read + Seek> FileReader<R> {
 
     /// Return the schema of the file
     pub fn schema(&self) -> SchemaRef {
-        self.schema.clone()
+        self.decoder.schema.clone()
     }
 
     /// Read a specific record batch
@@ -785,41 +908,7 @@ impl<R: Read + Seek> FileReader<R> {
 
         // read length
         let buffer = read_block(&mut self.reader, block)?;
-        let message = parse_message(&buffer)?;
-
-        // some old test data's footer metadata is not set, so we account for that
-        if self.metadata_version != MetadataVersion::V1
-            && message.version() != self.metadata_version
-        {
-            return Err(ArrowError::IpcError(
-                "Could not read IPC message as metadata versions mismatch".to_string(),
-            ));
-        }
-
-        match message.header_type() {
-            crate::MessageHeader::Schema => Err(ArrowError::IpcError(
-                "Not expecting a schema when messages are read".to_string(),
-            )),
-            crate::MessageHeader::RecordBatch => {
-                let batch = message.header_as_record_batch().ok_or_else(|| {
-                    ArrowError::IpcError("Unable to read IPC message as record batch".to_string())
-                })?;
-                // read the block that makes up the record batch into a buffer
-                read_record_batch(
-                    &buffer.slice(block.metaDataLength() as _),
-                    batch,
-                    self.schema(),
-                    &self.dictionaries_by_id,
-                    self.projection.as_ref().map(|x| x.0.as_ref()),
-                    &message.version(),
-                )
-                .map(Some)
-            }
-            crate::MessageHeader::NONE => Ok(None),
-            t => Err(ArrowError::InvalidArgumentError(format!(
-                "Reading types other than record batches not yet supported, unable to read {t:?}"
-            ))),
-        }
+        self.decoder.read_record_batch(block, &buffer)
     }
 
     /// Gets a reference to the underlying reader.
@@ -852,7 +941,7 @@ impl<R: Read + Seek> Iterator for FileReader<R> {
 
 impl<R: Read + Seek> RecordBatchReader for FileReader<R> {
     fn schema(&self) -> SchemaRef {
-        self.schema.clone()
+        self.schema()
     }
 }
 
diff --git a/arrow-ord/src/sort.rs b/arrow-ord/src/sort.rs
index 7d749da51..c25df3a48 100644
--- a/arrow-ord/src/sort.rs
+++ b/arrow-ord/src/sort.rs
@@ -790,7 +790,6 @@ mod tests {
     use half::f16;
     use rand::rngs::StdRng;
     use rand::{Rng, RngCore, SeedableRng};
-    use std::convert::TryFrom;
     use std::sync::Arc;
 
     fn create_decimal128_array(data: Vec<Option<i128>>) -> Decimal128Array {
@@ -972,19 +971,14 @@ mod tests {
 
         assert_eq!(sorted_dict, dict);
 
-        let sorted_strings = StringArray::try_from(
-            (0..sorted.len())
-                .map(|i| {
-                    if sorted.is_valid(i) {
-                        Some(sorted_dict.value(sorted_keys.value(i).as_usize()))
-                    } else {
-                        None
-                    }
-                })
-                .collect::<Vec<Option<&str>>>(),
-        )
-        .expect("Unable to create string array from dictionary");
-        let expected = StringArray::try_from(expected_data).expect("Unable to create string array");
+        let sorted_strings = StringArray::from_iter((0..sorted.len()).map(|i| {
+            if sorted.is_valid(i) {
+                Some(sorted_dict.value(sorted_keys.value(i).as_usize()))
+            } else {
+                None
+            }
+        }));
+        let expected = StringArray::from(expected_data);
 
         assert_eq!(sorted_strings, expected)
     }
diff --git a/arrow-schema/src/datatype.rs b/arrow-schema/src/datatype.rs
index 330ae5c9e..6276a99a4 100644
--- a/arrow-schema/src/datatype.rs
+++ b/arrow-schema/src/datatype.rs
@@ -23,14 +23,18 @@ use crate::{Field, FieldRef, Fields, UnionFields};
 /// The set of datatypes that are supported by this implementation of Apache Arrow.
 ///
 /// The Arrow specification on data types includes some more types.
-/// See also [`Schema.fbs`](https://github.com/apache/arrow/blob/master/format/Schema.fbs)
+/// See also [`Schema.fbs`](https://github.com/apache/arrow/blob/main/format/Schema.fbs)
 /// for Arrow's specification.
 ///
 /// The variants of this enum include primitive fixed size types as well as parametric or
 /// nested types.
-/// Currently the Rust implementation supports the following  nested types:
+/// Currently the Rust implementation supports the following nested types:
 ///  - `List<T>`
+///  - `LargeList<T>`
+///  - `FixedSizeList<T>`
 ///  - `Struct<T, U, V, ...>`
+///  - `Union<T, U, V, ...>`
+///  - `Map<K, V>`
 ///
 /// Nested types can themselves be nested within other arrays.
 /// For more information on these types please see
@@ -68,7 +72,7 @@ pub enum DataType {
     ///
     /// Time is measured as a Unix epoch, counting the seconds from
     /// 00:00:00.000 on 1 January 1970, excluding leap seconds,
-    /// as a 64-bit integer.
+    /// as a signed 64-bit integer.
     ///
     /// The time zone is a string indicating the name of a time zone, one of:
     ///
@@ -140,15 +144,17 @@ pub enum DataType {
     /// DataType::Timestamp(TimeUnit::Second, Some("string".to_string().into()));
     /// ```
     Timestamp(TimeUnit, Option<Arc<str>>),
-    /// A 32-bit date representing the elapsed time since UNIX epoch (1970-01-01)
+    /// A signed 32-bit date representing the elapsed time since UNIX epoch (1970-01-01)
     /// in days (32 bits).
     Date32,
-    /// A 64-bit date representing the elapsed time since UNIX epoch (1970-01-01)
+    /// A signed 64-bit date representing the elapsed time since UNIX epoch (1970-01-01)
     /// in milliseconds (64 bits). Values are evenly divisible by 86400000.
     Date64,
-    /// A 32-bit time representing the elapsed time since midnight in the unit of `TimeUnit`.
+    /// A signed 32-bit time representing the elapsed time since midnight in the unit of `TimeUnit`.
+    /// Must be either seconds or milliseconds.
     Time32(TimeUnit),
-    /// A 64-bit time representing the elapsed time since midnight in the unit of `TimeUnit`.
+    /// A signed 64-bit time representing the elapsed time since midnight in the unit of `TimeUnit`.
+    /// Must be either microseconds or nanoseconds.
     Time64(TimeUnit),
     /// Measure of elapsed time in either seconds, milliseconds, microseconds or nanoseconds.
     Duration(TimeUnit),
@@ -159,7 +165,7 @@ pub enum DataType {
     /// Opaque binary data of variable length.
     ///
     /// A single Binary array can store up to [`i32::MAX`] bytes
-    /// of binary data in total
+    /// of binary data in total.
     Binary,
     /// Opaque binary data of fixed size.
     /// Enum parameter specifies the number of bytes per value.
@@ -167,27 +173,27 @@ pub enum DataType {
     /// Opaque binary data of variable length and 64-bit offsets.
     ///
     /// A single LargeBinary array can store up to [`i64::MAX`] bytes
-    /// of binary data in total
+    /// of binary data in total.
     LargeBinary,
-    /// A variable-length string in Unicode with UTF-8 encoding
+    /// A variable-length string in Unicode with UTF-8 encoding.
     ///
     /// A single Utf8 array can store up to [`i32::MAX`] bytes
-    /// of string data in total
+    /// of string data in total.
     Utf8,
     /// A variable-length string in Unicode with UFT-8 encoding and 64-bit offsets.
     ///
     /// A single LargeUtf8 array can store up to [`i64::MAX`] bytes
-    /// of string data in total
+    /// of string data in total.
     LargeUtf8,
     /// A list of some logical data type with variable length.
     ///
-    /// A single List array can store up to [`i32::MAX`] elements in total
+    /// A single List array can store up to [`i32::MAX`] elements in total.
     List(FieldRef),
     /// A list of some logical data type with fixed length.
     FixedSizeList(FieldRef, i32),
     /// A list of some logical data type with variable length and 64-bit offsets.
     ///
-    /// A single LargeList array can store up to [`i64::MAX`] elements in total
+    /// A single LargeList array can store up to [`i64::MAX`] elements in total.
     LargeList(FieldRef),
     /// A nested datatype that contains a number of sub-fields.
     Struct(Fields),
@@ -573,9 +579,13 @@ impl DataType {
         }
     }
 
-    /// Create a List DataType default name is "item"
+    /// Create a [`DataType::List`] with elements of the specified type
+    /// and nullability, and conventionally named inner [`Field`] (`"item"`).
+    ///
+    /// To specify field level metadata, construct the inner [`Field`]
+    /// directly via [`Field::new`] or [`Field::new_list_field`].
     pub fn new_list(data_type: DataType, nullable: bool) -> Self {
-        DataType::List(Arc::new(Field::new("item", data_type, nullable)))
+        DataType::List(Arc::new(Field::new_list_field(data_type, nullable)))
     }
 }
 
diff --git a/arrow-schema/src/field.rs b/arrow-schema/src/field.rs
index 574c024bb..eebaf0e3c 100644
--- a/arrow-schema/src/field.rs
+++ b/arrow-schema/src/field.rs
@@ -117,7 +117,7 @@ impl Hash for Field {
 }
 
 impl Field {
-    /// Creates a new field
+    /// Creates a new field with the given name, type, and nullability
     pub fn new(name: impl Into<String>, data_type: DataType, nullable: bool) -> Self {
         Field {
             name: name.into(),
@@ -129,6 +129,24 @@ impl Field {
         }
     }
 
+    /// Creates a new `Field`` suitable for [`DataType::List`] and
+    /// [`DataType::LargeList`]
+    ///
+    /// While not required, this method follows the convention of naming the
+    /// `Field` `"item"`.
+    ///
+    /// # Example
+    /// ```
+    /// # use arrow_schema::{Field, DataType};
+    /// assert_eq!(
+    ///   Field::new("item", DataType::Int32, true),
+    ///   Field::new_list_field(DataType::Int32, true)
+    /// );
+    /// ```
+    pub fn new_list_field(data_type: DataType, nullable: bool) -> Self {
+        Self::new("item", data_type, nullable)
+    }
+
     /// Creates a new field that has additional dictionary information
     pub fn new_dict(
         name: impl Into<String>,
diff --git a/arrow-string/src/regexp.rs b/arrow-string/src/regexp.rs
index 25c712d20..5e539b91b 100644
--- a/arrow-string/src/regexp.rs
+++ b/arrow-string/src/regexp.rs
@@ -19,10 +19,11 @@
 //! expression of a \[Large\]StringArray
 
 use arrow_array::builder::{BooleanBufferBuilder, GenericStringBuilder, ListBuilder};
+use arrow_array::cast::AsArray;
 use arrow_array::*;
 use arrow_buffer::NullBuffer;
 use arrow_data::{ArrayData, ArrayDataBuilder};
-use arrow_schema::{ArrowError, DataType};
+use arrow_schema::{ArrowError, DataType, Field};
 use regex::Regex;
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -152,28 +153,7 @@ pub fn regexp_is_match_utf8_scalar<OffsetSize: OffsetSizeTrait>(
     Ok(BooleanArray::from(data))
 }
 
-/// Extract all groups matched by a regular expression for a given String array.
-///
-/// Modelled after the Postgres [regexp_match].
-///
-/// Returns a ListArray of [`GenericStringArray`] with each element containing the leftmost-first
-/// match of the corresponding index in `regex_array` to string in `array`
-///
-/// If there is no match, the list element is NULL.
-///
-/// If a match is found, and the pattern contains no capturing parenthesized subexpressions,
-/// then the list element is a single-element [`GenericStringArray`] containing the substring
-/// matching the whole pattern.
-///
-/// If a match is found, and the pattern contains capturing parenthesized subexpressions, then the
-/// list element is a [`GenericStringArray`] whose n'th element is the substring matching
-/// the n'th capturing parenthesized subexpression of the pattern.
-///
-/// The flags parameter is an optional text string containing zero or more single-letter flags
-/// that change the function's behavior.
-///
-/// [regexp_match]: https://www.postgresql.org/docs/current/functions-matching.html#FUNCTIONS-POSIX-REGEXP
-pub fn regexp_match<OffsetSize: OffsetSizeTrait>(
+fn regexp_array_match<OffsetSize: OffsetSizeTrait>(
     array: &GenericStringArray<OffsetSize>,
     regex_array: &GenericStringArray<OffsetSize>,
     flags_array: Option<&GenericStringArray<OffsetSize>>,
@@ -248,6 +228,179 @@ pub fn regexp_match<OffsetSize: OffsetSizeTrait>(
     Ok(Arc::new(list_builder.finish()))
 }
 
+fn get_scalar_pattern_flag<'a, OffsetSize: OffsetSizeTrait>(
+    regex_array: &'a dyn Array,
+    flag_array: Option<&'a dyn Array>,
+) -> (Option<&'a str>, Option<&'a str>) {
+    let regex = regex_array.as_string::<OffsetSize>();
+    let regex = regex.is_valid(0).then(|| regex.value(0));
+
+    if let Some(flag_array) = flag_array {
+        let flag = flag_array.as_string::<OffsetSize>();
+        (regex, flag.is_valid(0).then(|| flag.value(0)))
+    } else {
+        (regex, None)
+    }
+}
+
+fn regexp_scalar_match<OffsetSize: OffsetSizeTrait>(
+    array: &GenericStringArray<OffsetSize>,
+    regex: &Regex,
+) -> Result<ArrayRef, ArrowError> {
+    let builder: GenericStringBuilder<OffsetSize> = GenericStringBuilder::with_capacity(0, 0);
+    let mut list_builder = ListBuilder::new(builder);
+
+    array
+        .iter()
+        .map(|value| {
+            match value {
+                // Required for Postgres compatibility:
+                // SELECT regexp_match('foobarbequebaz', ''); = {""}
+                Some(_) if regex.as_str() == "" => {
+                    list_builder.values().append_value("");
+                    list_builder.append(true);
+                }
+                Some(value) => match regex.captures(value) {
+                    Some(caps) => {
+                        let mut iter = caps.iter();
+                        if caps.len() > 1 {
+                            iter.next();
+                        }
+                        for m in iter.flatten() {
+                            list_builder.values().append_value(m.as_str());
+                        }
+
+                        list_builder.append(true);
+                    }
+                    None => list_builder.append(false),
+                },
+                _ => list_builder.append(false),
+            }
+            Ok(())
+        })
+        .collect::<Result<Vec<()>, ArrowError>>()?;
+
+    Ok(Arc::new(list_builder.finish()))
+}
+
+/// Extract all groups matched by a regular expression for a given String array.
+///
+/// Modelled after the Postgres [regexp_match].
+///
+/// Returns a ListArray of [`GenericStringArray`] with each element containing the leftmost-first
+/// match of the corresponding index in `regex_array` to string in `array`
+///
+/// If there is no match, the list element is NULL.
+///
+/// If a match is found, and the pattern contains no capturing parenthesized subexpressions,
+/// then the list element is a single-element [`GenericStringArray`] containing the substring
+/// matching the whole pattern.
+///
+/// If a match is found, and the pattern contains capturing parenthesized subexpressions, then the
+/// list element is a [`GenericStringArray`] whose n'th element is the substring matching
+/// the n'th capturing parenthesized subexpression of the pattern.
+///
+/// The flags parameter is an optional text string containing zero or more single-letter flags
+/// that change the function's behavior.
+///
+/// [regexp_match]: https://www.postgresql.org/docs/current/functions-matching.html#FUNCTIONS-POSIX-REGEXP
+pub fn regexp_match(
+    array: &dyn Array,
+    regex_array: &dyn Datum,
+    flags_array: Option<&dyn Datum>,
+) -> Result<ArrayRef, ArrowError> {
+    let (rhs, is_rhs_scalar) = regex_array.get();
+
+    if array.data_type() != rhs.data_type() {
+        return Err(ArrowError::ComputeError(
+            "regexp_match() requires both array and pattern to be either Utf8 or LargeUtf8"
+                .to_string(),
+        ));
+    }
+
+    let (flags, is_flags_scalar) = match flags_array {
+        Some(flags) => {
+            let (flags, is_flags_scalar) = flags.get();
+            (Some(flags), Some(is_flags_scalar))
+        }
+        None => (None, None),
+    };
+
+    if is_flags_scalar.is_some() && is_rhs_scalar != is_flags_scalar.unwrap() {
+        return Err(ArrowError::ComputeError(
+            "regexp_match() requires both pattern and flags to be either scalar or array"
+                .to_string(),
+        ));
+    }
+
+    if flags_array.is_some() && rhs.data_type() != flags.unwrap().data_type() {
+        return Err(ArrowError::ComputeError(
+            "regexp_match() requires both pattern and flags to be either string or largestring"
+                .to_string(),
+        ));
+    }
+
+    if is_rhs_scalar {
+        // Regex and flag is scalars
+        let (regex, flag) = match rhs.data_type() {
+            DataType::Utf8 => get_scalar_pattern_flag::<i32>(rhs, flags),
+            DataType::LargeUtf8 => get_scalar_pattern_flag::<i64>(rhs, flags),
+            _ => {
+                return Err(ArrowError::ComputeError(
+                    "regexp_match() requires pattern to be either Utf8 or LargeUtf8".to_string(),
+                ));
+            }
+        };
+
+        if regex.is_none() {
+            return Ok(new_null_array(
+                &DataType::List(Arc::new(Field::new(
+                    "item",
+                    array.data_type().clone(),
+                    true,
+                ))),
+                array.len(),
+            ));
+        }
+
+        let regex = regex.unwrap();
+
+        let pattern = if let Some(flag) = flag {
+            format!("(?{flag}){regex}")
+        } else {
+            regex.to_string()
+        };
+
+        let re = Regex::new(pattern.as_str()).map_err(|e| {
+            ArrowError::ComputeError(format!("Regular expression did not compile: {e:?}"))
+        })?;
+
+        match array.data_type() {
+            DataType::Utf8 => regexp_scalar_match(array.as_string::<i32>(), &re),
+            DataType::LargeUtf8 => regexp_scalar_match(array.as_string::<i64>(), &re),
+            _ => Err(ArrowError::ComputeError(
+                "regexp_match() requires array to be either Utf8 or LargeUtf8".to_string(),
+            )),
+        }
+    } else {
+        match array.data_type() {
+            DataType::Utf8 => {
+                let regex_array = rhs.as_string();
+                let flags_array = flags.map(|flags| flags.as_string());
+                regexp_array_match(array.as_string::<i32>(), regex_array, flags_array)
+            }
+            DataType::LargeUtf8 => {
+                let regex_array = rhs.as_string();
+                let flags_array = flags.map(|flags| flags.as_string());
+                regexp_array_match(array.as_string::<i64>(), regex_array, flags_array)
+            }
+            _ => Err(ArrowError::ComputeError(
+                "regexp_match() requires array to be either Utf8 or LargeUtf8".to_string(),
+            )),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -304,6 +457,49 @@ mod tests {
         assert_eq!(&expected, result);
     }
 
+    #[test]
+    fn match_scalar_pattern() {
+        let values = vec![Some("abc-005-def"), Some("X-7-5"), Some("X545"), None];
+        let array = StringArray::from(values);
+        let pattern = Scalar::new(StringArray::from(vec![r"x.*-(\d*)-.*"; 1]));
+        let flags = Scalar::new(StringArray::from(vec!["i"; 1]));
+        let actual = regexp_match(&array, &pattern, Some(&flags)).unwrap();
+        let elem_builder: GenericStringBuilder<i32> = GenericStringBuilder::with_capacity(0, 0);
+        let mut expected_builder = ListBuilder::new(elem_builder);
+        expected_builder.append(false);
+        expected_builder.values().append_value("7");
+        expected_builder.append(true);
+        expected_builder.append(false);
+        expected_builder.append(false);
+        let expected = expected_builder.finish();
+        let result = actual.as_any().downcast_ref::<ListArray>().unwrap();
+        assert_eq!(&expected, result);
+
+        // No flag
+        let values = vec![Some("abc-005-def"), Some("x-7-5"), Some("X545"), None];
+        let array = StringArray::from(values);
+        let actual = regexp_match(&array, &pattern, None).unwrap();
+        let result = actual.as_any().downcast_ref::<ListArray>().unwrap();
+        assert_eq!(&expected, result);
+    }
+
+    #[test]
+    fn match_scalar_no_pattern() {
+        let values = vec![Some("abc-005-def"), Some("X-7-5"), Some("X545"), None];
+        let array = StringArray::from(values);
+        let pattern = Scalar::new(new_null_array(&DataType::Utf8, 1));
+        let actual = regexp_match(&array, &pattern, None).unwrap();
+        let elem_builder: GenericStringBuilder<i32> = GenericStringBuilder::with_capacity(0, 0);
+        let mut expected_builder = ListBuilder::new(elem_builder);
+        expected_builder.append(false);
+        expected_builder.append(false);
+        expected_builder.append(false);
+        expected_builder.append(false);
+        let expected = expected_builder.finish();
+        let result = actual.as_any().downcast_ref::<ListArray>().unwrap();
+        assert_eq!(&expected, result);
+    }
+
     #[test]
     fn test_single_group_not_skip_match() {
         let array = StringArray::from(vec![Some("foo"), Some("bar")]);
diff --git a/arrow/benches/regexp_kernels.rs b/arrow/benches/regexp_kernels.rs
index eb38ba678..d5ffbcb99 100644
--- a/arrow/benches/regexp_kernels.rs
+++ b/arrow/benches/regexp_kernels.rs
@@ -25,7 +25,7 @@ use arrow::array::*;
 use arrow::compute::kernels::regexp::*;
 use arrow::util::bench_util::*;
 
-fn bench_regexp(arr: &GenericStringArray<i32>, regex_array: &GenericStringArray<i32>) {
+fn bench_regexp(arr: &GenericStringArray<i32>, regex_array: &dyn Datum) {
     regexp_match(criterion::black_box(arr), regex_array, None).unwrap();
 }
 
@@ -38,6 +38,13 @@ fn add_benchmark(c: &mut Criterion) {
     let pattern = GenericStringArray::<i32>::from(pattern_values);
 
     c.bench_function("regexp", |b| b.iter(|| bench_regexp(&arr_string, &pattern)));
+
+    let pattern_values = vec![r".*-(\d*)-.*"];
+    let pattern = Scalar::new(GenericStringArray::<i32>::from(pattern_values));
+
+    c.bench_function("regexp scalar", |b| {
+        b.iter(|| bench_regexp(&arr_string, &pattern))
+    });
 }
 
 criterion_group!(benches, add_benchmark);
diff --git a/arrow/src/ffi_stream.rs b/arrow/src/ffi_stream.rs
index bbec71e88..06128a1c9 100644
--- a/arrow/src/ffi_stream.rs
+++ b/arrow/src/ffi_stream.rs
@@ -54,6 +54,7 @@
 //! }
 //! ```
 
+use arrow_schema::DataType;
 use std::ffi::CStr;
 use std::ptr::addr_of;
 use std::{
@@ -356,16 +357,10 @@ impl Iterator for ArrowArrayStreamReader {
                 return None;
             }
 
-            let schema_ref = self.schema();
-            // NOTE: this parses the FFI_ArrowSchema again on each iterator call;
-            // should probably use from_ffi_and_data_type() instead.
-            let schema = FFI_ArrowSchema::try_from(schema_ref.as_ref()).ok()?;
-
-            let data = unsafe { from_ffi(array, &schema) }.ok()?;
-
-            let record_batch = RecordBatch::from(StructArray::from(data));
-
-            Some(Ok(record_batch))
+            let result = unsafe {
+                from_ffi_and_data_type(array, DataType::Struct(self.schema().fields().clone()))
+            };
+            Some(result.map(|data| RecordBatch::from(StructArray::from(data))))
         } else {
             let last_error = self.get_stream_last_error();
             let err = ArrowError::CDataInterface(last_error.unwrap());
diff --git a/object_store/src/aws/builder.rs b/object_store/src/aws/builder.rs
index 5f7f1c945..542f17af3 100644
--- a/object_store/src/aws/builder.rs
+++ b/object_store/src/aws/builder.rs
@@ -41,9 +41,6 @@ static DEFAULT_METADATA_ENDPOINT: &str = "http://169.254.169.254";
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
 enum Error {
-    #[snafu(display("Missing region"))]
-    MissingRegion,
-
     #[snafu(display("Missing bucket name"))]
     MissingBucketName,
 
@@ -559,19 +556,25 @@ impl AmazonS3Builder {
         Ok(())
     }
 
-    /// Set the AWS Access Key (required)
+    /// Set the AWS Access Key
     pub fn with_access_key_id(mut self, access_key_id: impl Into<String>) -> Self {
         self.access_key_id = Some(access_key_id.into());
         self
     }
 
-    /// Set the AWS Secret Access Key (required)
+    /// Set the AWS Secret Access Key
     pub fn with_secret_access_key(mut self, secret_access_key: impl Into<String>) -> Self {
         self.secret_access_key = Some(secret_access_key.into());
         self
     }
 
-    /// Set the region (e.g. `us-east-1`) (required)
+    /// Set the AWS Session Token to use for requests
+    pub fn with_token(mut self, token: impl Into<String>) -> Self {
+        self.token = Some(token.into());
+        self
+    }
+
+    /// Set the region, defaults to `us-east-1`
     pub fn with_region(mut self, region: impl Into<String>) -> Self {
         self.region = Some(region.into());
         self
@@ -583,25 +586,24 @@ impl AmazonS3Builder {
         self
     }
 
-    /// Sets the endpoint for communicating with AWS S3. Default value
-    /// is based on region. The `endpoint` field should be consistent with
-    /// the field `virtual_hosted_style_request'.
+    /// Sets the endpoint for communicating with AWS S3, defaults to the [region endpoint]
     ///
     /// For example, this might be set to `"http://localhost:4566:`
     /// for testing against a localstack instance.
-    /// If `virtual_hosted_style_request` is set to true then `endpoint`
-    /// should have bucket name included.
+    ///
+    /// The `endpoint` field should be consistent with [`Self::with_virtual_hosted_style_request`],
+    /// i.e. if `virtual_hosted_style_request` is set to true then `endpoint`
+    /// should have the bucket name included.
+    ///
+    /// By default, only HTTPS schemes are enabled. To connect to an HTTP endpoint, enable
+    /// [`Self::with_allow_http`].
+    ///
+    /// [region endpoint]: https://docs.aws.amazon.com/general/latest/gr/s3.html
     pub fn with_endpoint(mut self, endpoint: impl Into<String>) -> Self {
         self.endpoint = Some(endpoint.into());
         self
     }
 
-    /// Set the token to use for requests (passed to underlying provider)
-    pub fn with_token(mut self, token: impl Into<String>) -> Self {
-        self.token = Some(token.into());
-        self
-    }
-
     /// Set the credential provider overriding any other options
     pub fn with_credentials(mut self, credentials: AwsCredentialProvider) -> Self {
         self.credentials = Some(credentials);
@@ -741,7 +743,7 @@ impl AmazonS3Builder {
         }
 
         let bucket = self.bucket_name.context(MissingBucketNameSnafu)?;
-        let region = self.region.context(MissingRegionSnafu)?;
+        let region = self.region.unwrap_or_else(|| "us-east-1".to_string());
         let checksum = self.checksum_algorithm.map(|x| x.get()).transpose()?;
         let copy_if_not_exists = self.copy_if_not_exists.map(|x| x.get()).transpose()?;
         let put_precondition = self.conditional_put.map(|x| x.get()).transpose()?;
@@ -950,6 +952,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn s3_default_region() {
+        let builder = AmazonS3Builder::new()
+            .with_bucket_name("foo")
+            .build()
+            .unwrap();
+        assert_eq!(builder.client.config.region, "us-east-1");
+    }
+
     #[test]
     fn s3_test_urls() {
         let mut builder = AmazonS3Builder::new();
diff --git a/object_store/src/aws/dynamo.rs b/object_store/src/aws/dynamo.rs
index ce1500bf4..f12a42137 100644
--- a/object_store/src/aws/dynamo.rs
+++ b/object_store/src/aws/dynamo.rs
@@ -17,7 +17,9 @@
 
 //! A DynamoDB based lock system
 
+use std::borrow::Cow;
 use std::collections::HashMap;
+use std::future::Future;
 use std::time::{Duration, Instant};
 
 use chrono::Utc;
@@ -61,16 +63,24 @@ const STORE: &str = "DynamoDB";
 ///
 /// The DynamoDB schema is as follows:
 ///
-/// * A string hash key named `"key"`
+/// * A string partition key named `"path"`
+/// * A string sort key named `"etag"`
 /// * A numeric [TTL] attribute named `"ttl"`
 /// * A numeric attribute named `"generation"`
 /// * A numeric attribute named `"timeout"`
 ///
-/// To perform a conditional operation on an object with a given `path` and `etag` (if exists),
+/// An appropriate DynamoDB table can be created with the CLI as follows:
+///
+/// ```bash
+/// $ aws dynamodb create-table --table-name <TABLE_NAME> --key-schema AttributeName=path,KeyType=HASH AttributeName=etag,KeyType=RANGE --attribute-definitions AttributeName=path,AttributeType=S AttributeName=etag,AttributeType=S
+/// $ aws dynamodb update-time-to-live --table-name <TABLE_NAME> --time-to-live-specification Enabled=true,AttributeName=ttl
+/// ```
+///
+/// To perform a conditional operation on an object with a given `path` and `etag` (`*` if creating),
 /// the commit protocol is as follows:
 ///
 /// 1. Perform HEAD request on `path` and error on precondition mismatch
-/// 2. Create record in DynamoDB with key `{path}#{etag}` with the configured timeout
+/// 2. Create record in DynamoDB with given `path` and `etag` with the configured timeout
 ///     1. On Success: Perform operation with the configured timeout
 ///     2. On Conflict:
 ///         1. Periodically re-perform HEAD request on `path` and error on precondition mismatch
@@ -154,6 +164,16 @@ impl DynamoCommit {
         self
     }
 
+    /// Parse [`DynamoCommit`] from a string
+    pub(crate) fn from_str(value: &str) -> Option<Self> {
+        Some(match value.split_once(':') {
+            Some((table_name, timeout)) => {
+                Self::new(table_name.trim().to_string()).with_timeout(timeout.parse().ok()?)
+            }
+            None => Self::new(value.trim().to_string()),
+        })
+    }
+
     /// Returns the name of the DynamoDB table.
     pub(crate) fn table_name(&self) -> &str {
         &self.table_name
@@ -165,23 +185,41 @@ impl DynamoCommit {
         from: &Path,
         to: &Path,
     ) -> Result<()> {
-        check_not_exists(client, to).await?;
+        self.conditional_op(client, to, None, || async {
+            client.copy_request(from, to).send().await?;
+            Ok(())
+        })
+        .await
+    }
+
+    #[allow(clippy::future_not_send)] // Generics confound this lint
+    pub(crate) async fn conditional_op<F, Fut, T>(
+        &self,
+        client: &S3Client,
+        to: &Path,
+        etag: Option<&str>,
+        op: F,
+    ) -> Result<T>
+    where
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Result<T, Error>>,
+    {
+        check_precondition(client, to, etag).await?;
 
         let mut previous_lease = None;
 
         loop {
             let existing = previous_lease.as_ref();
-            match self.try_lock(client, to.as_ref(), existing).await? {
+            match self.try_lock(client, to.as_ref(), etag, existing).await? {
                 TryLockResult::Ok(lease) => {
-                    let fut = client.copy_request(from, to).send();
                     let expiry = lease.acquire + lease.timeout;
-                    return match tokio::time::timeout_at(expiry.into(), fut).await {
-                        Ok(Ok(_)) => Ok(()),
-                        Ok(Err(e)) => Err(e.into()),
+                    return match tokio::time::timeout_at(expiry.into(), op()).await {
+                        Ok(Ok(v)) => Ok(v),
+                        Ok(Err(e)) => Err(e),
                         Err(_) => Err(Error::Generic {
                             store: "DynamoDB",
                             source: format!(
-                                "Failed to perform copy operation in {} milliseconds",
+                                "Failed to perform conditional operation in {} milliseconds",
                                 self.timeout
                             )
                             .into(),
@@ -193,7 +231,7 @@ impl DynamoCommit {
                     let expiry = conflict.timeout * self.max_clock_skew_rate;
                     loop {
                         interval.tick().await;
-                        check_not_exists(client, to).await?;
+                        check_precondition(client, to, etag).await?;
                         if conflict.acquire.elapsed() > expiry {
                             previous_lease = Some(conflict);
                             break;
@@ -205,8 +243,11 @@ impl DynamoCommit {
     }
 
     /// Retrieve a lock, returning an error if it doesn't exist
-    async fn get_lock(&self, s3: &S3Client, key: &str) -> Result<Lease> {
-        let key_attributes = [("key", AttributeValue::String(key))];
+    async fn get_lock(&self, s3: &S3Client, path: &str, etag: Option<&str>) -> Result<Lease> {
+        let key_attributes = [
+            ("path", AttributeValue::from(path)),
+            ("etag", AttributeValue::from(etag.unwrap_or("*"))),
+        ];
         let req = GetItem {
             table_name: &self.table_name,
             key: Map(&key_attributes),
@@ -216,7 +257,7 @@ impl DynamoCommit {
         let resp = self
             .request(s3, credential.as_deref(), "DynamoDB_20120810.GetItem", req)
             .await
-            .map_err(|e| e.error(STORE, key.to_string()))?;
+            .map_err(|e| e.error(STORE, path.to_string()))?;
 
         let body = resp.bytes().await.map_err(|e| Error::Generic {
             store: STORE,
@@ -230,7 +271,7 @@ impl DynamoCommit {
             })?;
 
         extract_lease(&response.item).ok_or_else(|| Error::NotFound {
-            path: key.into(),
+            path: path.into(),
             source: "DynamoDB GetItem returned no items".to_string().into(),
         })
     }
@@ -239,7 +280,8 @@ impl DynamoCommit {
     async fn try_lock(
         &self,
         s3: &S3Client,
-        key: &str,
+        path: &str,
+        etag: Option<&str>,
         existing: Option<&Lease>,
     ) -> Result<TryLockResult> {
         let attributes;
@@ -257,12 +299,13 @@ impl DynamoCommit {
 
         let ttl = (Utc::now() + self.ttl).timestamp();
         let items = [
-            ("key", AttributeValue::String(key)),
+            ("path", AttributeValue::from(path)),
+            ("etag", AttributeValue::from(etag.unwrap_or("*"))),
             ("generation", AttributeValue::Number(next_gen)),
             ("timeout", AttributeValue::Number(self.timeout)),
             ("ttl", AttributeValue::Number(ttl as _)),
         ];
-        let names = [("#pk", "key")];
+        let names = [("#pk", "path")];
 
         let req = PutItem {
             table_name: &self.table_name,
@@ -302,7 +345,9 @@ impl DynamoCommit {
                     // <https://aws.amazon.com/about-aws/whats-new/2023/06/amazon-dynamodb-cost-failed-conditional-writes/>
                     // <https://repost.aws/questions/QUNfADrK4RT6WHe61RzTK8aw/dynamodblocal-support-for-returnvaluesonconditioncheckfailure-for-single-write-operations>
                     // <https://github.com/localstack/localstack/issues/9040>
-                    None => Ok(TryLockResult::Conflict(self.get_lock(s3, key).await?)),
+                    None => Ok(TryLockResult::Conflict(
+                        self.get_lock(s3, path, etag).await?,
+                    )),
                 },
                 _ => Err(Error::Generic {
                     store: STORE,
@@ -347,19 +392,37 @@ enum TryLockResult {
     Conflict(Lease),
 }
 
-/// Returns an [`Error::AlreadyExists`] if `path` exists
-async fn check_not_exists(client: &S3Client, path: &Path) -> Result<()> {
+/// Validates that `path` has the given `etag` or doesn't exist if `None`
+async fn check_precondition(client: &S3Client, path: &Path, etag: Option<&str>) -> Result<()> {
     let options = GetOptions {
         head: true,
         ..Default::default()
     };
-    match client.get_opts(path, options).await {
-        Ok(_) => Err(Error::AlreadyExists {
-            path: path.to_string(),
-            source: "Already Exists".to_string().into(),
-        }),
-        Err(Error::NotFound { .. }) => Ok(()),
-        Err(e) => Err(e),
+
+    match etag {
+        Some(expected) => match client.get_opts(path, options).await {
+            Ok(r) => match r.meta.e_tag {
+                Some(actual) if expected == actual => Ok(()),
+                actual => Err(Error::Precondition {
+                    path: path.to_string(),
+                    source: format!("{} does not match {expected}", actual.unwrap_or_default())
+                        .into(),
+                }),
+            },
+            Err(Error::NotFound { .. }) => Err(Error::Precondition {
+                path: path.to_string(),
+                source: format!("Object at location {path} not found").into(),
+            }),
+            Err(e) => Err(e),
+        },
+        None => match client.get_opts(path, options).await {
+            Ok(_) => Err(Error::AlreadyExists {
+                path: path.to_string(),
+                source: "Already Exists".to_string().into(),
+            }),
+            Err(Error::NotFound { .. }) => Ok(()),
+            Err(e) => Err(e),
+        },
     }
 }
 
@@ -493,11 +556,17 @@ impl<'a, K: Serialize, V: Serialize> Serialize for Map<'a, K, V> {
 #[derive(Debug, Serialize, Deserialize)]
 enum AttributeValue<'a> {
     #[serde(rename = "S")]
-    String(&'a str),
+    String(Cow<'a, str>),
     #[serde(rename = "N", with = "number")]
     Number(u64),
 }
 
+impl<'a> From<&'a str> for AttributeValue<'a> {
+    fn from(value: &'a str) -> Self {
+        Self::String(Cow::Borrowed(value))
+    }
+}
+
 /// Numbers are serialized as strings
 mod number {
     use serde::{Deserialize, Deserializer, Serializer};
@@ -518,10 +587,11 @@ pub(crate) use tests::integration_test;
 
 #[cfg(test)]
 mod tests {
-
     use super::*;
     use crate::aws::AmazonS3;
     use crate::ObjectStore;
+    use rand::distributions::Alphanumeric;
+    use rand::{thread_rng, Rng};
 
     #[test]
     fn test_attribute_serde() {
@@ -544,24 +614,43 @@ mod tests {
         let _ = integration.delete(&dst).await; // Delete if present
 
         // Create a lock if not already exists
-        let existing = match d.try_lock(client, dst.as_ref(), None).await.unwrap() {
+        let existing = match d.try_lock(client, dst.as_ref(), None, None).await.unwrap() {
             TryLockResult::Conflict(l) => l,
             TryLockResult::Ok(l) => l,
         };
 
         // Should not be able to acquire a lock again
-        let r = d.try_lock(client, dst.as_ref(), None).await;
+        let r = d.try_lock(client, dst.as_ref(), None, None).await;
         assert!(matches!(r, Ok(TryLockResult::Conflict(_))));
 
         // But should still be able to reclaim lock and perform copy
         d.copy_if_not_exists(client, &src, &dst).await.unwrap();
 
-        match d.try_lock(client, dst.as_ref(), None).await.unwrap() {
+        match d.try_lock(client, dst.as_ref(), None, None).await.unwrap() {
             TryLockResult::Conflict(new) => {
                 // Should have incremented generation to do so
                 assert_eq!(new.generation, existing.generation + 1);
             }
             _ => panic!("Should conflict"),
         }
+
+        let rng = thread_rng();
+        let etag = String::from_utf8(rng.sample_iter(Alphanumeric).take(32).collect()).unwrap();
+        let t = Some(etag.as_str());
+
+        let l = match d.try_lock(client, dst.as_ref(), t, None).await.unwrap() {
+            TryLockResult::Ok(l) => l,
+            _ => panic!("should not conflict"),
+        };
+
+        match d.try_lock(client, dst.as_ref(), t, None).await.unwrap() {
+            TryLockResult::Conflict(c) => assert_eq!(l.generation, c.generation),
+            _ => panic!("should conflict"),
+        }
+
+        match d.try_lock(client, dst.as_ref(), t, Some(&l)).await.unwrap() {
+            TryLockResult::Ok(new) => assert_eq!(new.generation, l.generation + 1),
+            _ => panic!("should not conflict"),
+        }
     }
 }
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 75b43d448..d167c78e4 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -187,12 +187,26 @@ impl ObjectStore for AmazonS3 {
                     r => r,
                 }
             }
-            (PutMode::Update(v), Some(S3ConditionalPut::ETagMatch)) => {
+            (PutMode::Create, Some(S3ConditionalPut::Dynamo(d))) => {
+                d.conditional_op(&self.client, location, None, move || request.do_put())
+                    .await
+            }
+            (PutMode::Update(v), Some(put)) => {
                 let etag = v.e_tag.ok_or_else(|| Error::Generic {
                     store: STORE,
                     source: "ETag required for conditional put".to_string().into(),
                 })?;
-                request.header(&IF_MATCH, etag.as_str()).do_put().await
+                match put {
+                    S3ConditionalPut::ETagMatch => {
+                        request.header(&IF_MATCH, etag.as_str()).do_put().await
+                    }
+                    S3ConditionalPut::Dynamo(d) => {
+                        d.conditional_op(&self.client, location, Some(&etag), move || {
+                            request.do_put()
+                        })
+                        .await
+                    }
+                }
             }
         }
     }
@@ -372,6 +386,7 @@ mod tests {
         rename_and_copy(&integration).await;
         stream_get(&integration).await;
         multipart(&integration, &integration).await;
+        signing(&integration).await;
 
         tagging(&integration, !config.disable_tagging, |p| {
             let client = Arc::clone(&integration.client);
diff --git a/object_store/src/aws/precondition.rs b/object_store/src/aws/precondition.rs
index 83d45db82..ad9e21537 100644
--- a/object_store/src/aws/precondition.rs
+++ b/object_store/src/aws/precondition.rs
@@ -48,7 +48,7 @@ pub enum S3CopyIfNotExists {
     HeaderWithStatus(String, String, reqwest::StatusCode),
     /// The name of a DynamoDB table to use for coordination
     ///
-    /// Encoded as either `dynamodb:<TABLE_NAME>` or `dynamodb:<TABLE_NAME>:<TIMEOUT_MILLIS>`
+    /// Encoded as either `dynamo:<TABLE_NAME>` or `dynamo:<TABLE_NAME>:<TIMEOUT_MILLIS>`
     /// ignoring whitespace. The default timeout is used if not specified
     ///
     /// See [`DynamoCommit`] for more information
@@ -88,12 +88,7 @@ impl S3CopyIfNotExists {
                     code,
                 ))
             }
-            "dynamo" => Some(Self::Dynamo(match value.split_once(':') {
-                Some((table_name, timeout)) => DynamoCommit::new(table_name.trim().to_string())
-                    .with_timeout(timeout.parse().ok()?),
-                None => DynamoCommit::new(value.trim().to_string()),
-            })),
-
+            "dynamo" => Some(Self::Dynamo(DynamoCommit::from_str(value)?)),
             _ => None,
         }
     }
@@ -111,7 +106,7 @@ impl Parse for S3CopyIfNotExists {
 /// Configure how to provide conditional put support for [`AmazonS3`].
 ///
 /// [`AmazonS3`]: super::AmazonS3
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Eq, PartialEq)]
 #[allow(missing_copy_implementations)]
 #[non_exhaustive]
 pub enum S3ConditionalPut {
@@ -122,12 +117,23 @@ pub enum S3ConditionalPut {
     ///
     /// [HTTP precondition]: https://datatracker.ietf.org/doc/html/rfc9110#name-preconditions
     ETagMatch,
+
+    /// The name of a DynamoDB table to use for coordination
+    ///
+    /// Encoded as either `dynamo:<TABLE_NAME>` or `dynamo:<TABLE_NAME>:<TIMEOUT_MILLIS>`
+    /// ignoring whitespace. The default timeout is used if not specified
+    ///
+    /// See [`DynamoCommit`] for more information
+    ///
+    /// This will use the same region, credentials and endpoint as configured for S3
+    Dynamo(DynamoCommit),
 }
 
 impl std::fmt::Display for S3ConditionalPut {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::ETagMatch => write!(f, "etag"),
+            Self::Dynamo(lock) => write!(f, "dynamo: {}", lock.table_name()),
         }
     }
 }
@@ -136,7 +142,10 @@ impl S3ConditionalPut {
     fn from_str(s: &str) -> Option<Self> {
         match s.trim() {
             "etag" => Some(Self::ETagMatch),
-            _ => None,
+            trimmed => match trimmed.split_once(':')? {
+                ("dynamo", s) => Some(Self::Dynamo(DynamoCommit::from_str(s)?)),
+                _ => None,
+            },
         }
     }
 }
@@ -153,6 +162,7 @@ impl Parse for S3ConditionalPut {
 #[cfg(test)]
 mod tests {
     use super::S3CopyIfNotExists;
+    use crate::aws::{DynamoCommit, S3ConditionalPut};
 
     #[test]
     fn parse_s3_copy_if_not_exists_header() {
@@ -177,6 +187,24 @@ mod tests {
         assert_eq!(expected, S3CopyIfNotExists::from_str(input));
     }
 
+    #[test]
+    fn parse_s3_copy_if_not_exists_dynamo() {
+        let input = "dynamo: table:100";
+        let expected = Some(S3CopyIfNotExists::Dynamo(
+            DynamoCommit::new("table".into()).with_timeout(100),
+        ));
+        assert_eq!(expected, S3CopyIfNotExists::from_str(input));
+    }
+
+    #[test]
+    fn parse_s3_condition_put_dynamo() {
+        let input = "dynamo: table:1300";
+        let expected = Some(S3ConditionalPut::Dynamo(
+            DynamoCommit::new("table".into()).with_timeout(1300),
+        ));
+        assert_eq!(expected, S3ConditionalPut::from_str(input));
+    }
+
     #[test]
     fn parse_s3_copy_if_not_exists_header_whitespace_invariant() {
         let expected = Some(S3CopyIfNotExists::Header(
diff --git a/object_store/src/azure/builder.rs b/object_store/src/azure/builder.rs
index 2de0a7c05..905fa5216 100644
--- a/object_store/src/azure/builder.rs
+++ b/object_store/src/azure/builder.rs
@@ -687,6 +687,9 @@ impl MicrosoftAzureBuilder {
     /// Override the endpoint used to communicate with blob storage
     ///
     /// Defaults to `https://{account}.blob.core.windows.net`
+    ///
+    /// By default, only HTTPS schemes are enabled. To connect to an HTTP endpoint, enable
+    /// [`Self::with_allow_http`].
     pub fn with_endpoint(mut self, endpoint: String) -> Self {
         self.endpoint = Some(endpoint);
         self
diff --git a/object_store/src/azure/client.rs b/object_store/src/azure/client.rs
index 3c71e69da..865e0a1a9 100644
--- a/object_store/src/azure/client.rs
+++ b/object_store/src/azure/client.rs
@@ -46,6 +46,7 @@ use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 use url::Url;
 
 const VERSION_HEADER: &str = "x-ms-version-id";
@@ -101,6 +102,18 @@ pub(crate) enum Error {
 
     #[snafu(display("ETag required for conditional update"))]
     MissingETag,
+
+    #[snafu(display("Error requesting user delegation key: {}", source))]
+    DelegationKeyRequest { source: crate::client::retry::Error },
+
+    #[snafu(display("Error getting user delegation key response body: {}", source))]
+    DelegationKeyResponseBody { source: reqwest::Error },
+
+    #[snafu(display("Got invalid user delegation key response: {}", source))]
+    DelegationKeyResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Generating SAS keys with SAS tokens auth is not supported"))]
+    SASforSASNotSupported,
 }
 
 impl From<Error> for crate::Error {
@@ -324,6 +337,78 @@ impl AzureClient {
         Ok(())
     }
 
+    /// Make a Get User Delegation Key request
+    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/get-user-delegation-key>
+    async fn get_user_delegation_key(
+        &self,
+        start: &DateTime<Utc>,
+        end: &DateTime<Utc>,
+    ) -> Result<UserDelegationKey> {
+        let credential = self.get_credential().await?;
+        let url = self.config.service.clone();
+
+        let start = start.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
+        let expiry = end.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
+
+        let mut body = String::new();
+        body.push_str("<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<KeyInfo>\n");
+        body.push_str(&format!(
+            "\t<Start>{start}</Start>\n\t<Expiry>{expiry}</Expiry>\n"
+        ));
+        body.push_str("</KeyInfo>");
+
+        let response = self
+            .client
+            .request(Method::POST, url)
+            .body(body)
+            .query(&[("restype", "service"), ("comp", "userdelegationkey")])
+            .with_azure_authorization(&credential, &self.config.account)
+            .send_retry(&self.config.retry_config)
+            .await
+            .context(DelegationKeyRequestSnafu)?
+            .bytes()
+            .await
+            .context(DelegationKeyResponseBodySnafu)?;
+
+        let response: UserDelegationKey =
+            quick_xml::de::from_reader(response.reader()).context(DelegationKeyResponseSnafu)?;
+
+        Ok(response)
+    }
+
+    /// Creat an AzureSigner for generating SAS tokens (pre-signed urls).
+    ///
+    /// Depending on the type of credential, this will either use the account key or a user delegation key.
+    /// Since delegation keys are acquired ad-hoc, the signer aloows for signing multiple urls with the same key.
+    pub async fn signer(&self, expires_in: Duration) -> Result<AzureSigner> {
+        let credential = self.get_credential().await?;
+        let signed_start = chrono::Utc::now();
+        let signed_expiry = signed_start + expires_in;
+        match credential.as_ref() {
+            AzureCredential::BearerToken(_) => {
+                let key = self
+                    .get_user_delegation_key(&signed_start, &signed_expiry)
+                    .await?;
+                let signing_key = AzureAccessKey::try_new(&key.value)?;
+                Ok(AzureSigner::new(
+                    signing_key,
+                    self.config.account.clone(),
+                    signed_start,
+                    signed_expiry,
+                    Some(key),
+                ))
+            }
+            AzureCredential::AccessKey(key) => Ok(AzureSigner::new(
+                key.to_owned(),
+                self.config.account.clone(),
+                signed_start,
+                signed_expiry,
+                None,
+            )),
+            _ => Err(Error::SASforSASNotSupported.into()),
+        }
+    }
+
     #[cfg(test)]
     pub async fn get_blob_tagging(&self, path: &Path) -> Result<Response> {
         let credential = self.get_credential().await?;
@@ -600,6 +685,18 @@ impl BlockList {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub(crate) struct UserDelegationKey {
+    pub signed_oid: String,
+    pub signed_tid: String,
+    pub signed_start: String,
+    pub signed_expiry: String,
+    pub signed_service: String,
+    pub signed_version: String,
+    pub value: String,
+}
+
 #[cfg(test)]
 mod tests {
     use bytes::Bytes;
@@ -757,8 +854,7 @@ mod tests {
     <NextMarker/>
 </EnumerationResults>";
 
-        let mut _list_blobs_response_internal: ListResultInternal =
-            quick_xml::de::from_str(S).unwrap();
+        let _list_blobs_response_internal: ListResultInternal = quick_xml::de::from_str(S).unwrap();
     }
 
     #[test]
@@ -778,4 +874,21 @@ mod tests {
 
         assert_eq!(res, S)
     }
+
+    #[test]
+    fn test_delegated_key_response() {
+        const S: &str = r#"<?xml version="1.0" encoding="utf-8"?>
+<UserDelegationKey>
+    <SignedOid>String containing a GUID value</SignedOid>
+    <SignedTid>String containing a GUID value</SignedTid>
+    <SignedStart>String formatted as ISO date</SignedStart>
+    <SignedExpiry>String formatted as ISO date</SignedExpiry>
+    <SignedService>b</SignedService>
+    <SignedVersion>String specifying REST api version to use to create the user delegation key</SignedVersion>
+    <Value>String containing the user delegation key</Value>
+</UserDelegationKey>"#;
+
+        let _delegated_key_response_internal: UserDelegationKey =
+            quick_xml::de::from_str(S).unwrap();
+    }
 }
diff --git a/object_store/src/azure/credential.rs b/object_store/src/azure/credential.rs
index 2b8788d33..bfbbde826 100644
--- a/object_store/src/azure/credential.rs
+++ b/object_store/src/azure/credential.rs
@@ -24,26 +24,27 @@ use crate::RetryConfig;
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
-use chrono::{DateTime, Utc};
-use reqwest::header::ACCEPT;
-use reqwest::{
-    header::{
-        HeaderMap, HeaderName, HeaderValue, AUTHORIZATION, CONTENT_ENCODING, CONTENT_LANGUAGE,
-        CONTENT_LENGTH, CONTENT_TYPE, DATE, IF_MATCH, IF_MODIFIED_SINCE, IF_NONE_MATCH,
-        IF_UNMODIFIED_SINCE, RANGE,
-    },
-    Client, Method, RequestBuilder,
+use chrono::{DateTime, SecondsFormat, Utc};
+use reqwest::header::{
+    HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, CONTENT_ENCODING, CONTENT_LANGUAGE,
+    CONTENT_LENGTH, CONTENT_TYPE, DATE, IF_MATCH, IF_MODIFIED_SINCE, IF_NONE_MATCH,
+    IF_UNMODIFIED_SINCE, RANGE,
 };
+use reqwest::{Client, Method, Request, RequestBuilder};
 use serde::Deserialize;
 use snafu::{ResultExt, Snafu};
 use std::borrow::Cow;
+use std::collections::HashMap;
+use std::fmt::Debug;
 use std::process::Command;
 use std::str;
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime};
 use url::Url;
 
-static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+use super::client::UserDelegationKey;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2023-11-03");
 static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
 pub(crate) static BLOB_TYPE: HeaderName = HeaderName::from_static("x-ms-blob-type");
 pub(crate) static DELETE_SNAPSHOTS: HeaderName = HeaderName::from_static("x-ms-delete-snapshots");
@@ -83,6 +84,9 @@ pub enum Error {
 
     #[snafu(display("Failed to parse azure cli response: {source}"))]
     AzureCliResponse { source: serde_json::Error },
+
+    #[snafu(display("Generating SAS keys with SAS tokens auth is not supported"))]
+    SASforSASNotSupported,
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -97,7 +101,7 @@ impl From<Error> for crate::Error {
 }
 
 /// A shared Azure Storage Account Key
-#[derive(Debug, Eq, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq)]
 pub struct AzureAccessKey(Vec<u8>);
 
 impl AzureAccessKey {
@@ -137,33 +141,86 @@ pub mod authority_hosts {
     pub const AZURE_PUBLIC_CLOUD: &str = "https://login.microsoftonline.com";
 }
 
-pub(crate) trait CredentialExt {
-    /// Apply authorization to requests against azure storage accounts
-    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-requests-to-azure-storage>
-    fn with_azure_authorization(self, credential: &AzureCredential, account: &str) -> Self;
+pub(crate) struct AzureSigner {
+    signing_key: AzureAccessKey,
+    start: DateTime<Utc>,
+    end: DateTime<Utc>,
+    account: String,
+    delegation_key: Option<UserDelegationKey>,
 }
 
-impl CredentialExt for RequestBuilder {
-    fn with_azure_authorization(mut self, credential: &AzureCredential, account: &str) -> Self {
+impl AzureSigner {
+    pub fn new(
+        signing_key: AzureAccessKey,
+        account: String,
+        start: DateTime<Utc>,
+        end: DateTime<Utc>,
+        delegation_key: Option<UserDelegationKey>,
+    ) -> Self {
+        Self {
+            signing_key,
+            account,
+            start,
+            end,
+            delegation_key,
+        }
+    }
+
+    pub fn sign(&self, method: &Method, url: &mut Url) -> Result<()> {
+        let (str_to_sign, query_pairs) = match &self.delegation_key {
+            Some(delegation_key) => string_to_sign_user_delegation_sas(
+                url,
+                method,
+                &self.account,
+                &self.start,
+                &self.end,
+                delegation_key,
+            ),
+            None => string_to_sign_service_sas(url, method, &self.account, &self.start, &self.end),
+        };
+        let auth = hmac_sha256(&self.signing_key.0, str_to_sign);
+        url.query_pairs_mut().extend_pairs(query_pairs);
+        url.query_pairs_mut()
+            .append_pair("sig", BASE64_STANDARD.encode(auth).as_str());
+        Ok(())
+    }
+}
+
+/// Authorize a [`Request`] with an [`AzureAuthorizer`]
+#[derive(Debug)]
+pub struct AzureAuthorizer<'a> {
+    credential: &'a AzureCredential,
+    account: &'a str,
+}
+
+impl<'a> AzureAuthorizer<'a> {
+    /// Create a new [`AzureAuthorizer`]
+    pub fn new(credential: &'a AzureCredential, account: &'a str) -> Self {
+        AzureAuthorizer {
+            credential,
+            account,
+        }
+    }
+
+    /// Authorize `request`
+    pub fn authorize(&self, request: &mut Request) {
         // rfc2822 string should never contain illegal characters
         let date = Utc::now();
         let date_str = date.format(RFC1123_FMT).to_string();
         // we formatted the data string ourselves, so unwrapping should be fine
         let date_val = HeaderValue::from_str(&date_str).unwrap();
-        self = self
-            .header(DATE, &date_val)
-            .header(&VERSION, &AZURE_VERSION);
+        request.headers_mut().insert(DATE, date_val);
+        request
+            .headers_mut()
+            .insert(&VERSION, AZURE_VERSION.clone());
 
-        match credential {
+        match self.credential {
             AzureCredential::AccessKey(key) => {
-                let (client, request) = self.build_split();
-                let mut request = request.expect("request valid");
-
                 let signature = generate_authorization(
                     request.headers(),
                     request.url(),
                     request.method(),
-                    account,
+                    self.account,
                     key,
                 );
 
@@ -173,15 +230,40 @@ impl CredentialExt for RequestBuilder {
                     AUTHORIZATION,
                     HeaderValue::from_str(signature.as_str()).unwrap(),
                 );
-
-                Self::from_parts(client, request)
             }
-            AzureCredential::BearerToken(token) => self.bearer_auth(token),
-            AzureCredential::SASToken(query_pairs) => self.query(&query_pairs),
+            AzureCredential::BearerToken(token) => {
+                request.headers_mut().append(
+                    AUTHORIZATION,
+                    HeaderValue::from_str(format!("Bearer {}", token).as_str()).unwrap(),
+                );
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                request
+                    .url_mut()
+                    .query_pairs_mut()
+                    .extend_pairs(query_pairs);
+            }
         }
     }
 }
 
+pub(crate) trait CredentialExt {
+    /// Apply authorization to requests against azure storage accounts
+    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-requests-to-azure-storage>
+    fn with_azure_authorization(self, credential: &AzureCredential, account: &str) -> Self;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(self, credential: &AzureCredential, account: &str) -> Self {
+        let (client, request) = self.build_split();
+        let mut request = request.expect("request valid");
+
+        AzureAuthorizer::new(credential, account).authorize(&mut request);
+
+        Self::from_parts(client, request)
+    }
+}
+
 /// Generate signed key for authorization via access keys
 /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key>
 fn generate_authorization(
@@ -205,6 +287,152 @@ fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
         .unwrap_or_default()
 }
 
+fn string_to_sign_sas(
+    u: &Url,
+    method: &Method,
+    account: &str,
+    start: &DateTime<Utc>,
+    end: &DateTime<Utc>,
+) -> (String, String, String, String, String) {
+    // NOTE: for now only blob signing is supported.
+    let signed_resource = "b".to_string();
+
+    // https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas#permissions-for-a-directory-container-or-blob
+    let signed_permissions = match *method {
+        // read and list permissions
+        Method::GET => match signed_resource.as_str() {
+            "c" => "rl",
+            "b" => "r",
+            _ => unreachable!(),
+        },
+        // write permissions (also allows crating a new blob in a sub-key)
+        Method::PUT => "w",
+        // delete permissions
+        Method::DELETE => "d",
+        // other methods are not used in any of the current operations
+        _ => "",
+    }
+    .to_string();
+    let signed_start = start.to_rfc3339_opts(SecondsFormat::Secs, true);
+    let signed_expiry = end.to_rfc3339_opts(SecondsFormat::Secs, true);
+    let canonicalized_resource = if u.host_str().unwrap_or_default().contains(account) {
+        format!("/blob/{}{}", account, u.path())
+    } else {
+        // NOTE: in case of the emulator, the account name is not part of the host
+        //      but the path starts with the account name
+        format!("/blob{}", u.path())
+    };
+
+    (
+        signed_resource,
+        signed_permissions,
+        signed_start,
+        signed_expiry,
+        canonicalized_resource,
+    )
+}
+
+/// Create a string to be signed for authorization via [service sas].
+///
+/// [service sas]: https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas#version-2020-12-06-and-later
+fn string_to_sign_service_sas(
+    u: &Url,
+    method: &Method,
+    account: &str,
+    start: &DateTime<Utc>,
+    end: &DateTime<Utc>,
+) -> (String, HashMap<&'static str, String>) {
+    let (signed_resource, signed_permissions, signed_start, signed_expiry, canonicalized_resource) =
+        string_to_sign_sas(u, method, account, start, end);
+
+    let string_to_sign = format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}",
+        signed_permissions,
+        signed_start,
+        signed_expiry,
+        canonicalized_resource,
+        "",                               // signed identifier
+        "",                               // signed ip
+        "",                               // signed protocol
+        &AZURE_VERSION.to_str().unwrap(), // signed version
+        signed_resource,                  // signed resource
+        "",                               // signed snapshot time
+        "",                               // signed encryption scope
+        "",                               // rscc - response header: Cache-Control
+        "",                               // rscd - response header: Content-Disposition
+        "",                               // rsce - response header: Content-Encoding
+        "",                               // rscl - response header: Content-Language
+        "",                               // rsct - response header: Content-Type
+    );
+
+    let mut pairs = HashMap::new();
+    pairs.insert("sv", AZURE_VERSION.to_str().unwrap().to_string());
+    pairs.insert("sp", signed_permissions);
+    pairs.insert("st", signed_start);
+    pairs.insert("se", signed_expiry);
+    pairs.insert("sr", signed_resource);
+
+    (string_to_sign, pairs)
+}
+
+/// Create a string to be signed for authorization via [user delegation sas].
+///
+/// [user delegation sas]: https://learn.microsoft.com/en-us/rest/api/storageservices/create-user-delegation-sas#version-2020-12-06-and-later
+fn string_to_sign_user_delegation_sas(
+    u: &Url,
+    method: &Method,
+    account: &str,
+    start: &DateTime<Utc>,
+    end: &DateTime<Utc>,
+    delegation_key: &UserDelegationKey,
+) -> (String, HashMap<&'static str, String>) {
+    let (signed_resource, signed_permissions, signed_start, signed_expiry, canonicalized_resource) =
+        string_to_sign_sas(u, method, account, start, end);
+
+    let string_to_sign = format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}",
+        signed_permissions,
+        signed_start,
+        signed_expiry,
+        canonicalized_resource,
+        delegation_key.signed_oid,        // signed key object id
+        delegation_key.signed_tid,        // signed key tenant id
+        delegation_key.signed_start,      // signed key start
+        delegation_key.signed_expiry,     // signed key expiry
+        delegation_key.signed_service,    // signed key service
+        delegation_key.signed_version,    // signed key version
+        "",                               // signed authorized user object id
+        "",                               // signed unauthorized user object id
+        "",                               // signed correlation id
+        "",                               // signed ip
+        "",                               // signed protocol
+        &AZURE_VERSION.to_str().unwrap(), // signed version
+        signed_resource,                  // signed resource
+        "",                               // signed snapshot time
+        "",                               // signed encryption scope
+        "",                               // rscc - response header: Cache-Control
+        "",                               // rscd - response header: Content-Disposition
+        "",                               // rsce - response header: Content-Encoding
+        "",                               // rscl - response header: Content-Language
+        "",                               // rsct - response header: Content-Type
+    );
+
+    let mut pairs = HashMap::new();
+    pairs.insert("sv", AZURE_VERSION.to_str().unwrap().to_string());
+    pairs.insert("sp", signed_permissions);
+    pairs.insert("st", signed_start);
+    pairs.insert("se", signed_expiry);
+    pairs.insert("sr", signed_resource);
+    pairs.insert("skoid", delegation_key.signed_oid.clone());
+    pairs.insert("sktid", delegation_key.signed_tid.clone());
+    pairs.insert("skt", delegation_key.signed_start.clone());
+    pairs.insert("ske", delegation_key.signed_expiry.clone());
+    pairs.insert("sks", delegation_key.signed_service.clone());
+    pairs.insert("skv", delegation_key.signed_version.clone());
+
+    (string_to_sign, pairs)
+}
+
 /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-signature-string>
 fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
     // content length must only be specified if != 0
@@ -232,7 +460,7 @@ fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> Str
         add_if_exists(h, &IF_UNMODIFIED_SINCE),
         add_if_exists(h, &RANGE),
         canonicalize_header(h),
-        canonicalized_resource(account, u)
+        canonicalize_resource(account, u)
     )
 }
 
@@ -257,7 +485,7 @@ fn canonicalize_header(headers: &HeaderMap) -> String {
 }
 
 /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-canonicalized-resource-string>
-fn canonicalized_resource(account: &str, uri: &Url) -> String {
+fn canonicalize_resource(account: &str, uri: &Url) -> String {
     let mut can_res: String = String::new();
     can_res.push('/');
     can_res.push_str(account);
@@ -681,14 +909,15 @@ impl CredentialProvider for AzureCliCredential {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
-    use crate::client::mock_server::MockServer;
     use futures::executor::block_on;
     use hyper::body::to_bytes;
     use hyper::{Body, Response};
     use reqwest::{Client, Method};
     use tempfile::NamedTempFile;
 
+    use super::*;
+    use crate::client::mock_server::MockServer;
+
     #[tokio::test]
     async fn test_managed_identity() {
         let server = MockServer::new();
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index af0a4cefa..712b7a36c 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -27,22 +27,26 @@
 //! a way to drop old blocks. Instead unused blocks are automatically cleaned up
 //! after 7 days.
 use crate::{
-    multipart::{PartId, PutPart, WriteMultiPart},
+    multipart::{MultiPartStore, PartId, PutPart, WriteMultiPart},
     path::Path,
+    signer::Signer,
     GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, PutOptions, PutResult,
     Result,
 };
 use async_trait::async_trait;
 use bytes::Bytes;
 use futures::stream::BoxStream;
+use reqwest::Method;
 use std::fmt::Debug;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::io::AsyncWrite;
+use url::Url;
 
 use crate::client::get::GetClientExt;
 use crate::client::list::ListClientExt;
 use crate::client::CredentialProvider;
-pub use credential::authority_hosts;
+pub use credential::{authority_hosts, AzureAccessKey, AzureAuthorizer};
 
 mod builder;
 mod client;
@@ -50,7 +54,6 @@ mod credential;
 
 /// [`CredentialProvider`] for [`MicrosoftAzure`]
 pub type AzureCredentialProvider = Arc<dyn CredentialProvider<Credential = AzureCredential>>;
-use crate::multipart::MultiPartStore;
 pub use builder::{AzureConfigKey, MicrosoftAzureBuilder};
 pub use credential::AzureCredential;
 
@@ -67,6 +70,11 @@ impl MicrosoftAzure {
     pub fn credentials(&self) -> &AzureCredentialProvider {
         &self.client.config().credentials
     }
+
+    /// Create a full URL to the resource specified by `path` with this instance's configuration.
+    fn path_url(&self, path: &Path) -> url::Url {
+        self.client.config().path_url(path)
+    }
 }
 
 impl std::fmt::Display for MicrosoftAzure {
@@ -128,6 +136,62 @@ impl ObjectStore for MicrosoftAzure {
     }
 }
 
+#[async_trait]
+impl Signer for MicrosoftAzure {
+    /// Create a URL containing the relevant [Service SAS] query parameters that authorize a request
+    /// via `method` to the resource at `path` valid for the duration specified in `expires_in`.
+    ///
+    /// [Service SAS]: https://learn.microsoft.com/en-us/rest/api/storageservices/create-service-sas
+    ///
+    /// # Example
+    ///
+    /// This example returns a URL that will enable a user to upload a file to
+    /// "some-folder/some-file.txt" in the next hour.
+    ///
+    /// ```
+    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// # use object_store::{azure::MicrosoftAzureBuilder, path::Path, signer::Signer};
+    /// # use reqwest::Method;
+    /// # use std::time::Duration;
+    /// #
+    /// let azure = MicrosoftAzureBuilder::new()
+    ///     .with_account("my-account")
+    ///     .with_access_key("my-access-key")
+    ///     .with_container_name("my-container")
+    ///     .build()?;
+    ///
+    /// let url = azure.signed_url(
+    ///     Method::PUT,
+    ///     &Path::from("some-folder/some-file.txt"),
+    ///     Duration::from_secs(60 * 60)
+    /// ).await?;
+    /// #     Ok(())
+    /// # }
+    /// ```
+    async fn signed_url(&self, method: Method, path: &Path, expires_in: Duration) -> Result<Url> {
+        let mut url = self.path_url(path);
+        let signer = self.client.signer(expires_in).await?;
+        signer.sign(&method, &mut url)?;
+        Ok(url)
+    }
+
+    async fn signed_urls(
+        &self,
+        method: Method,
+        paths: &[Path],
+        expires_in: Duration,
+    ) -> Result<Vec<Url>> {
+        let mut urls = Vec::with_capacity(paths.len());
+        let signer = self.client.signer(expires_in).await?;
+        for path in paths {
+            let mut url = self.path_url(path);
+            signer.sign(&method, &mut url)?;
+            urls.push(url);
+        }
+        Ok(urls)
+    }
+}
+
 /// Relevant docs: <https://azure.github.io/Storage/docs/application-and-user-data/basics/azure-blob-storage-upload-apis/>
 /// In Azure Blob Store, parts are "blocks"
 /// put_multipart_part -> PUT block
@@ -202,6 +266,7 @@ mod tests {
         stream_get(&integration).await;
         put_opts(&integration, true).await;
         multipart(&integration, &integration).await;
+        signing(&integration).await;
 
         let validate = !integration.client.config().disable_tagging;
         tagging(&integration, validate, |p| {
@@ -211,6 +276,38 @@ mod tests {
         .await
     }
 
+    #[ignore = "Used for manual testing against a real storage account."]
+    #[tokio::test]
+    async fn test_user_delegation_key() {
+        let account = std::env::var("AZURE_ACCOUNT_NAME").unwrap();
+        let container = std::env::var("AZURE_CONTAINER_NAME").unwrap();
+        let client_id = std::env::var("AZURE_CLIENT_ID").unwrap();
+        let client_secret = std::env::var("AZURE_CLIENT_SECRET").unwrap();
+        let tenant_id = std::env::var("AZURE_TENANT_ID").unwrap();
+        let integration = MicrosoftAzureBuilder::new()
+            .with_account(account)
+            .with_container_name(container)
+            .with_client_id(client_id)
+            .with_client_secret(client_secret)
+            .with_tenant_id(&tenant_id)
+            .build()
+            .unwrap();
+
+        let data = Bytes::from("hello world");
+        let path = Path::from("file.txt");
+        integration.put(&path, data.clone()).await.unwrap();
+
+        let signed = integration
+            .signed_url(Method::GET, &path, Duration::from_secs(60))
+            .await
+            .unwrap();
+
+        let resp = reqwest::get(signed).await.unwrap();
+        let loaded = resp.bytes().await.unwrap();
+
+        assert_eq!(data, loaded);
+    }
+
     #[test]
     fn azure_test_config_get_value() {
         let azure_client_id = "object_store:fake_access_key_id".to_string();
diff --git a/object_store/src/client/get.rs b/object_store/src/client/get.rs
index 5f9cac9b4..b7e7f24b2 100644
--- a/object_store/src/client/get.rs
+++ b/object_store/src/client/get.rs
@@ -15,13 +15,18 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::ops::Range;
+
 use crate::client::header::{header_meta, HeaderConfig};
 use crate::path::Path;
-use crate::{Error, GetOptions, GetResult};
-use crate::{GetResultPayload, Result};
+use crate::{Error, GetOptions, GetResult, GetResultPayload, Result};
 use async_trait::async_trait;
 use futures::{StreamExt, TryStreamExt};
+use hyper::header::CONTENT_RANGE;
+use hyper::StatusCode;
+use reqwest::header::ToStrError;
 use reqwest::Response;
+use snafu::{ensure, OptionExt, ResultExt, Snafu};
 
 /// A client that can perform a get request
 #[async_trait]
@@ -45,25 +50,221 @@ impl<T: GetClient> GetClientExt for T {
     async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
         let range = options.range.clone();
         let response = self.get_request(location, options).await?;
-        let meta = header_meta(location, response.headers(), T::HEADER_CONFIG).map_err(|e| {
-            Error::Generic {
-                store: T::STORE,
-                source: Box::new(e),
-            }
-        })?;
-
-        let stream = response
-            .bytes_stream()
-            .map_err(|source| Error::Generic {
-                store: T::STORE,
-                source: Box::new(source),
-            })
-            .boxed();
-
-        Ok(GetResult {
-            range: range.unwrap_or(0..meta.size),
-            payload: GetResultPayload::Stream(stream),
-            meta,
+        get_result::<T>(location, range, response).map_err(|e| crate::Error::Generic {
+            store: T::STORE,
+            source: Box::new(e),
+        })
+    }
+}
+
+struct ContentRange {
+    /// The range of the object returned
+    range: Range<usize>,
+    /// The total size of the object being requested
+    size: usize,
+}
+
+impl ContentRange {
+    /// Parse a content range of the form `bytes <range-start>-<range-end>/<size>`
+    ///
+    /// <https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Range>
+    fn from_str(s: &str) -> Option<Self> {
+        let rem = s.trim().strip_prefix("bytes ")?;
+        let (range, size) = rem.split_once('/')?;
+        let size = size.parse().ok()?;
+
+        let (start_s, end_s) = range.split_once('-')?;
+
+        let start = start_s.parse().ok()?;
+        let end: usize = end_s.parse().ok()?;
+
+        Some(Self {
+            size,
+            range: start..end + 1,
         })
     }
 }
+
+/// A specialized `Error` for get-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum GetResultError {
+    #[snafu(context(false))]
+    Header {
+        source: crate::client::header::Error,
+    },
+
+    #[snafu(display("Received non-partial response when range requested"))]
+    NotPartial,
+
+    #[snafu(display("Content-Range header not present in partial response"))]
+    NoContentRange,
+
+    #[snafu(display("Failed to parse value for CONTENT_RANGE header: \"{value}\""))]
+    ParseContentRange { value: String },
+
+    #[snafu(display("Content-Range header contained non UTF-8 characters"))]
+    InvalidContentRange { source: ToStrError },
+
+    #[snafu(display("Requested {expected:?}, got {actual:?}"))]
+    UnexpectedRange {
+        expected: Range<usize>,
+        actual: Range<usize>,
+    },
+}
+
+fn get_result<T: GetClient>(
+    location: &Path,
+    range: Option<Range<usize>>,
+    response: Response,
+) -> Result<GetResult, GetResultError> {
+    let mut meta = header_meta(location, response.headers(), T::HEADER_CONFIG)?;
+
+    // ensure that we receive the range we asked for
+    let range = if let Some(expected) = range {
+        ensure!(
+            response.status() == StatusCode::PARTIAL_CONTENT,
+            NotPartialSnafu
+        );
+        let val = response
+            .headers()
+            .get(CONTENT_RANGE)
+            .context(NoContentRangeSnafu)?;
+
+        let value = val.to_str().context(InvalidContentRangeSnafu)?;
+        let value = ContentRange::from_str(value).context(ParseContentRangeSnafu { value })?;
+        let actual = value.range;
+
+        ensure!(
+            actual == expected,
+            UnexpectedRangeSnafu { expected, actual }
+        );
+
+        // Update size to reflect full size of object (#5272)
+        meta.size = value.size;
+        actual
+    } else {
+        0..meta.size
+    };
+
+    let stream = response
+        .bytes_stream()
+        .map_err(|source| Error::Generic {
+            store: T::STORE,
+            source: Box::new(source),
+        })
+        .boxed();
+
+    Ok(GetResult {
+        range,
+        meta,
+        payload: GetResultPayload::Stream(stream),
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use hyper::http;
+    use hyper::http::header::*;
+
+    struct TestClient {}
+
+    #[async_trait]
+    impl GetClient for TestClient {
+        const STORE: &'static str = "TEST";
+
+        const HEADER_CONFIG: HeaderConfig = HeaderConfig {
+            etag_required: false,
+            last_modified_required: false,
+            version_header: None,
+        };
+
+        async fn get_request(&self, _: &Path, _: GetOptions) -> Result<Response> {
+            unimplemented!()
+        }
+    }
+
+    fn make_response(
+        object_size: usize,
+        range: Option<Range<usize>>,
+        status: StatusCode,
+        content_range: Option<&str>,
+    ) -> Response {
+        let mut builder = http::Response::builder();
+        if let Some(range) = content_range {
+            builder = builder.header(CONTENT_RANGE, range);
+        }
+
+        let body = match range {
+            Some(range) => vec![0_u8; range.end - range.start],
+            None => vec![0_u8; object_size],
+        };
+
+        builder
+            .status(status)
+            .header(CONTENT_LENGTH, object_size)
+            .body(body)
+            .unwrap()
+            .into()
+    }
+
+    #[tokio::test]
+    async fn test_get_result() {
+        let path = Path::from("test");
+
+        let resp = make_response(12, None, StatusCode::OK, None);
+        let res = get_result::<TestClient>(&path, None, resp).unwrap();
+        assert_eq!(res.meta.size, 12);
+        assert_eq!(res.range, 0..12);
+        let bytes = res.bytes().await.unwrap();
+        assert_eq!(bytes.len(), 12);
+
+        let resp = make_response(
+            12,
+            Some(2..3),
+            StatusCode::PARTIAL_CONTENT,
+            Some("bytes 2-2/12"),
+        );
+        let res = get_result::<TestClient>(&path, Some(2..3), resp).unwrap();
+        assert_eq!(res.meta.size, 12);
+        assert_eq!(res.range, 2..3);
+        let bytes = res.bytes().await.unwrap();
+        assert_eq!(bytes.len(), 1);
+
+        let resp = make_response(12, Some(2..3), StatusCode::OK, None);
+        let err = get_result::<TestClient>(&path, Some(2..3), resp).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "Received non-partial response when range requested"
+        );
+
+        let resp = make_response(
+            12,
+            Some(2..3),
+            StatusCode::PARTIAL_CONTENT,
+            Some("bytes 2-3/12"),
+        );
+        let err = get_result::<TestClient>(&path, Some(2..3), resp).unwrap_err();
+        assert_eq!(err.to_string(), "Requested 2..3, got 2..4");
+
+        let resp = make_response(
+            12,
+            Some(2..3),
+            StatusCode::PARTIAL_CONTENT,
+            Some("bytes 2-2/*"),
+        );
+        let err = get_result::<TestClient>(&path, Some(2..3), resp).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "Failed to parse value for CONTENT_RANGE header: \"bytes 2-2/*\""
+        );
+
+        let resp = make_response(12, Some(2..3), StatusCode::PARTIAL_CONTENT, None);
+        let err = get_result::<TestClient>(&path, Some(2..3), resp).unwrap_err();
+        assert_eq!(
+            err.to_string(),
+            "Content-Range header not present in partial response"
+        );
+    }
+}
diff --git a/object_store/src/client/mock_server.rs b/object_store/src/client/mock_server.rs
index 36c6b650c..70b856186 100644
--- a/object_store/src/client/mock_server.rs
+++ b/object_store/src/client/mock_server.rs
@@ -15,17 +15,20 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use futures::future::BoxFuture;
+use futures::FutureExt;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Request, Response, Server};
 use parking_lot::Mutex;
 use std::collections::VecDeque;
 use std::convert::Infallible;
+use std::future::Future;
 use std::net::SocketAddr;
 use std::sync::Arc;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 
-pub type ResponseFn = Box<dyn FnOnce(Request<Body>) -> Response<Body> + Send>;
+pub type ResponseFn = Box<dyn FnOnce(Request<Body>) -> BoxFuture<'static, Response<Body>> + Send>;
 
 /// A mock server
 pub struct MockServer {
@@ -46,9 +49,10 @@ impl MockServer {
             async move {
                 Ok::<_, Infallible>(service_fn(move |req| {
                     let r = Arc::clone(&r);
+                    let next = r.lock().pop_front();
                     async move {
-                        Ok::<_, Infallible>(match r.lock().pop_front() {
-                            Some(r) => r(req),
+                        Ok::<_, Infallible>(match next {
+                            Some(r) => r(req).await,
                             None => Response::new(Body::from("Hello World")),
                         })
                     }
@@ -93,7 +97,16 @@ impl MockServer {
     where
         F: FnOnce(Request<Body>) -> Response<Body> + Send + 'static,
     {
-        self.responses.lock().push_back(Box::new(f))
+        let f = Box::new(|req| async move { f(req) }.boxed());
+        self.responses.lock().push_back(f)
+    }
+
+    pub fn push_async_fn<F, Fut>(&self, f: F)
+    where
+        F: FnOnce(Request<Body>) -> Fut + Send + 'static,
+        Fut: Future<Output = Response<Body>> + Send + 'static,
+    {
+        self.responses.lock().push_back(Box::new(|r| f(r).boxed()))
     }
 
     /// Shutdown the mock server
diff --git a/object_store/src/client/retry.rs b/object_store/src/client/retry.rs
index 08b9a74e1..9d21867d8 100644
--- a/object_store/src/client/retry.rs
+++ b/object_store/src/client/retry.rs
@@ -119,11 +119,19 @@ impl From<Error> for std::io::Error {
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
-/// Contains the configuration for how to respond to server errors
+/// The configuration for how to respond to request errors
 ///
-/// By default they will be retried up to some limit, using exponential
+/// The following categories of error will be retried:
+///
+/// * 5xx server errors
+/// * Connection errors
+/// * Dropped connections
+/// * Timeouts for [safe] / read-only requests
+///
+/// Requests will be retried up to some limit, using exponential
 /// backoff with jitter. See [`BackoffConfig`] for more information
 ///
+/// [safe]: https://datatracker.ietf.org/doc/html/rfc7231#section-4.2.1
 #[derive(Debug, Clone)]
 pub struct RetryConfig {
     /// The backoff configuration
@@ -173,13 +181,16 @@ impl RetryExt for reqwest::RequestBuilder {
         let max_retries = config.max_retries;
         let retry_timeout = config.retry_timeout;
 
+        let (client, req) = self.build_split();
+        let req = req.expect("request must be valid");
+
         async move {
             let mut retries = 0;
             let now = Instant::now();
 
             loop {
-                let s = self.try_clone().expect("request body must be cloneable");
-                match s.send().await {
+                let s = req.try_clone().expect("request body must be cloneable");
+                match client.execute(s).await {
                     Ok(r) => match r.error_for_status_ref() {
                         Ok(_) if r.status().is_success() => return Ok(r),
                         Ok(r) if r.status() == StatusCode::NOT_MODIFIED => {
@@ -242,7 +253,9 @@ impl RetryExt for reqwest::RequestBuilder {
                     Err(e) =>
                     {
                         let mut do_retry = false;
-                        if let Some(source) = e.source() {
+                        if req.method().is_safe() && e.is_timeout() {
+                            do_retry = true
+                        } else if let Some(source) = e.source() {
                             if let Some(e) = source.downcast_ref::<hyper::Error>() {
                                 if e.is_connect() || e.is_closed() || e.is_incomplete_message() {
                                     do_retry = true;
@@ -294,7 +307,11 @@ mod tests {
             retry_timeout: Duration::from_secs(1000),
         };
 
-        let client = Client::new();
+        let client = Client::builder()
+            .timeout(Duration::from_millis(100))
+            .build()
+            .unwrap();
+
         let do_request = || client.request(Method::GET, mock.url()).send_retry(&retry);
 
         // Simple request should work
@@ -419,7 +436,7 @@ mod tests {
 
         let e = do_request().await.unwrap_err().to_string();
         assert!(
-            e.contains("Error after 2 retries in") && 
+            e.contains("Error after 2 retries in") &&
             e.contains("max_retries:2, retry_timeout:1000s, source:HTTP status server error (502 Bad Gateway) for url"),
             "{e}"
         );
@@ -442,6 +459,25 @@ mod tests {
             "{e}"
         );
 
+        // Retries on client timeout
+        mock.push_async_fn(|_| async move {
+            tokio::time::sleep(Duration::from_secs(10)).await;
+            panic!()
+        });
+        do_request().await.unwrap();
+
+        // Does not retry PUT request
+        mock.push_async_fn(|_| async move {
+            tokio::time::sleep(Duration::from_secs(10)).await;
+            panic!()
+        });
+        let res = client.request(Method::PUT, mock.url()).send_retry(&retry);
+        let e = res.await.unwrap_err().to_string();
+        assert!(
+            e.contains("Error after 0 retries in") && e.contains("operation timed out"),
+            "{e}"
+        );
+
         // Shutdown
         mock.shutdown().await
     }
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index 3a841667f..8fc47b2c5 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -1233,6 +1233,7 @@ mod tests {
     use crate::test_util::flatten_list_stream;
     use chrono::TimeZone;
     use futures::stream::FuturesUnordered;
+    use rand::distributions::Alphanumeric;
     use rand::{thread_rng, Rng};
     use std::future::Future;
     use tokio::io::AsyncWriteExt;
@@ -1303,12 +1304,23 @@ mod tests {
         let range = 3..7;
         let range_result = storage.get_range(&location, range.clone()).await;
 
+        let bytes = range_result.unwrap();
+        assert_eq!(bytes, expected_data.slice(range.clone()));
+
+        let opts = GetOptions {
+            range: Some(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;
 
-        let bytes = range_result.unwrap();
-        assert_eq!(bytes, expected_data.slice(range));
-
         // Should be a non-fatal error
         out_of_range_result.unwrap_err();
 
@@ -1715,8 +1727,15 @@ mod tests {
     }
 
     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("put_opts");
+        let path = Path::from(format!("put_opts_{suffix}"));
         let v1 = storage
             .put_opts(&path, "a".into(), PutMode::Create.into())
             .await
@@ -1768,7 +1787,7 @@ mod tests {
         const NUM_WORKERS: usize = 5;
         const NUM_INCREMENTS: usize = 10;
 
-        let path = Path::from("RACE");
+        let path = Path::from(format!("RACE-{suffix}"));
         let mut futures: FuturesUnordered<_> = (0..NUM_WORKERS)
             .map(|_| async {
                 for _ in 0..NUM_INCREMENTS {
@@ -1886,7 +1905,7 @@ mod tests {
 
         // We can abort an in-progress write
         let (upload_id, mut writer) = storage.put_multipart(&location).await.unwrap();
-        if let Some(chunk) = data.get(0) {
+        if let Some(chunk) = data.first() {
             writer.write_all(chunk).await.unwrap();
             let _ = writer.write(chunk).await.unwrap();
         }
@@ -2115,6 +2134,29 @@ mod tests {
         assert_eq!(meta.size, chunk_size * 2);
     }
 
+    #[cfg(any(feature = "azure", feature = "aws"))]
+    pub(crate) async fn signing<T>(integration: &T)
+    where
+        T: ObjectStore + crate::signer::Signer,
+    {
+        use reqwest::Method;
+        use std::time::Duration;
+
+        let data = Bytes::from("hello world");
+        let path = Path::from("file.txt");
+        integration.put(&path, data.clone()).await.unwrap();
+
+        let signed = integration
+            .signed_url(Method::GET, &path, Duration::from_secs(60))
+            .await
+            .unwrap();
+
+        let resp = reqwest::get(signed).await.unwrap();
+        let loaded = resp.bytes().await.unwrap();
+
+        assert_eq!(data, loaded);
+    }
+
     #[cfg(any(feature = "aws", feature = "azure"))]
     pub(crate) async fn tagging<F, Fut>(storage: &dyn ObjectStore, validate: bool, get_tags: F)
     where
diff --git a/object_store/src/signer.rs b/object_store/src/signer.rs
index ed92e2879..da55c689a 100644
--- a/object_store/src/signer.rs
+++ b/object_store/src/signer.rs
@@ -31,4 +31,20 @@ pub trait Signer: Send + Sync + fmt::Debug + 'static {
     /// implementation's credentials such that the URL can be handed to something that doesn't have
     /// access to the object store's credentials, to allow limited access to the object store.
     async fn signed_url(&self, method: Method, path: &Path, expires_in: Duration) -> Result<Url>;
+
+    /// Generate signed urls for multiple paths.
+    ///
+    /// See [`Signer::signed_url`] for more details.
+    async fn signed_urls(
+        &self,
+        method: Method,
+        paths: &[Path],
+        expires_in: Duration,
+    ) -> Result<Vec<Url>> {
+        let mut urls = Vec::with_capacity(paths.len());
+        for path in paths {
+            urls.push(self.signed_url(method.clone(), path, expires_in).await?);
+        }
+        Ok(urls)
+    }
 }
diff --git a/parquet/src/arrow/arrow_writer/levels.rs b/parquet/src/arrow/arrow_writer/levels.rs
index df37665ce..4fa04aa15 100644
--- a/parquet/src/arrow/arrow_writer/levels.rs
+++ b/parquet/src/arrow/arrow_writer/levels.rs
@@ -1058,7 +1058,7 @@ mod tests {
 
         assert_eq!(levels.len(), 1);
 
-        let list_level = levels.get(0).unwrap();
+        let list_level = &levels[0];
 
         let expected_level = ArrayLevels {
             def_levels: Some(vec![0, 3, 3, 3]),
@@ -1150,7 +1150,7 @@ mod tests {
         assert_eq!(levels.len(), 5);
 
         // test "a" levels
-        let list_level = levels.get(0).unwrap();
+        let list_level = &levels[0];
 
         let expected_level = ArrayLevels {
             def_levels: None,
@@ -1295,7 +1295,7 @@ mod tests {
         let map = batch.column(0).as_map();
 
         // test key levels
-        let list_level = levels.get(0).unwrap();
+        let list_level = &levels[0];
 
         let expected_level = ArrayLevels {
             def_levels: Some(vec![1; 7]),
@@ -1477,9 +1477,10 @@ mod tests {
         let field_a2 = Arc::new(Field::new("integers", a2.data_type().clone(), true));
 
         let nulls = Buffer::from([0b00110111]);
-        let struct_a =
-            Arc::new(StructArray::try_from((vec![(field_a1, a1), (field_a2, a2)], nulls)).unwrap())
-                as ArrayRef;
+        let struct_a = Arc::new(StructArray::from((
+            vec![(field_a1, a1), (field_a2, a2)],
+            nulls,
+        ))) as ArrayRef;
 
         let offsets = Buffer::from_iter([0_i32, 0, 2, 2, 3, 5, 5]);
         let nulls = Buffer::from([0b00111100]);
@@ -1566,7 +1567,7 @@ mod tests {
 
         assert_eq!(levels.len(), 1);
 
-        let list_level = levels.get(0).unwrap();
+        let list_level = &levels[0];
 
         let expected_level = ArrayLevels {
             def_levels: Some(vec![0, 0, 3, 3]),
@@ -1756,7 +1757,7 @@ mod tests {
 
         assert_eq!(levels.len(), 1);
 
-        let list_level = levels.get(0).unwrap();
+        let list_level = &levels[0];
 
         let expected_level = ArrayLevels {
             def_levels: Some(vec![1, 0, 1]),
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index 597386b33..e993cb4c1 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -2584,14 +2584,8 @@ mod tests {
         }
 
         // page location
-        assert_eq!(
-            0,
-            offset_index.page_locations.get(0).unwrap().first_row_index
-        );
-        assert_eq!(
-            4,
-            offset_index.page_locations.get(1).unwrap().first_row_index
-        );
+        assert_eq!(0, offset_index.page_locations[0].first_row_index);
+        assert_eq!(4, offset_index.page_locations[1].first_row_index);
     }
 
     /// Verify min/max value truncation in the column index works as expected
@@ -2632,8 +2626,8 @@ mod tests {
             assert_eq!(stats.null_count(), 0);
             assert_eq!(stats.distinct_count(), None);
             if let Statistics::FixedLenByteArray(stats) = stats {
-                let column_index_min_value = column_index.min_values.get(0).unwrap();
-                let column_index_max_value = column_index.max_values.get(0).unwrap();
+                let column_index_min_value = &column_index.min_values[0];
+                let column_index_max_value = &column_index.max_values[0];
 
                 // Column index stats are truncated, while the column chunk's aren't.
                 assert_ne!(stats.min_bytes(), column_index_min_value.as_slice());
@@ -2699,8 +2693,8 @@ mod tests {
             assert_eq!(stats.null_count(), 0);
             assert_eq!(stats.distinct_count(), None);
             if let Statistics::FixedLenByteArray(_stats) = stats {
-                let column_index_min_value = column_index.min_values.get(0).unwrap();
-                let column_index_max_value = column_index.max_values.get(0).unwrap();
+                let column_index_min_value = &column_index.min_values[0];
+                let column_index_max_value = &column_index.max_values[0];
 
                 assert_eq!(column_index_min_value.len(), 1);
                 assert_eq!(column_index_max_value.len(), 1);
diff --git a/parquet/src/file/serialized_reader.rs b/parquet/src/file/serialized_reader.rs
index 468433f31..f960b1830 100644
--- a/parquet/src/file/serialized_reader.rs
+++ b/parquet/src/file/serialized_reader.rs
@@ -1116,16 +1116,13 @@ mod tests {
 
         assert_eq!(metadata.len(), 3);
 
-        assert_eq!(metadata.get(0).unwrap().key, "parquet.proto.descriptor");
+        assert_eq!(metadata[0].key, "parquet.proto.descriptor");
 
-        assert_eq!(metadata.get(1).unwrap().key, "writer.model.name");
-        assert_eq!(metadata.get(1).unwrap().value, Some("protobuf".to_owned()));
+        assert_eq!(metadata[1].key, "writer.model.name");
+        assert_eq!(metadata[1].value, Some("protobuf".to_owned()));
 
-        assert_eq!(metadata.get(2).unwrap().key, "parquet.proto.class");
-        assert_eq!(
-            metadata.get(2).unwrap().value,
-            Some("foo.baz.Foobaz$Event".to_owned())
-        );
+        assert_eq!(metadata[2].key, "parquet.proto.class");
+        assert_eq!(metadata[2].value, Some("foo.baz.Foobaz$Event".to_owned()));
     }
 
     #[test]
@@ -1141,7 +1138,7 @@ mod tests {
         assert_eq!(col0_metadata.bloom_filter_offset().unwrap(), 192);
 
         // test page encoding stats
-        let page_encoding_stats = col0_metadata.page_encoding_stats().unwrap().get(0).unwrap();
+        let page_encoding_stats = &col0_metadata.page_encoding_stats().unwrap()[0];
 
         assert_eq!(page_encoding_stats.page_type, basic::PageType::DATA_PAGE);
         assert_eq!(page_encoding_stats.encoding, Encoding::PLAIN);
diff --git a/parquet/src/file/writer.rs b/parquet/src/file/writer.rs
index 3534bab81..e0721a28c 100644
--- a/parquet/src/file/writer.rs
+++ b/parquet/src/file/writer.rs
@@ -1024,8 +1024,7 @@ mod tests {
         // ARROW-11803: Test that the converted and logical types have been populated
         let fields = reader.metadata().file_metadata().schema().get_fields();
         assert_eq!(fields.len(), 1);
-        let read_field = fields.get(0).unwrap();
-        assert_eq!(read_field, &field);
+        assert_eq!(fields[0], field);
     }
 
     #[test]


Test patch:
diff --git a/arrow-integration-test/src/lib.rs b/arrow-integration-test/src/lib.rs
index 7b797aa07..a312148dc 100644
--- a/arrow-integration-test/src/lib.rs
+++ b/arrow-integration-test/src/lib.rs
@@ -666,8 +666,7 @@ pub fn array_from_json(
         DataType::List(child_field) => {
             let null_buf = create_null_buf(&json_col);
             let children = json_col.children.clone().unwrap();
-            let child_array =
-                array_from_json(child_field, children.get(0).unwrap().clone(), dictionaries)?;
+            let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
             let offsets: Vec<i32> = json_col
                 .offset
                 .unwrap()
@@ -687,8 +686,7 @@ pub fn array_from_json(
         DataType::LargeList(child_field) => {
             let null_buf = create_null_buf(&json_col);
             let children = json_col.children.clone().unwrap();
-            let child_array =
-                array_from_json(child_field, children.get(0).unwrap().clone(), dictionaries)?;
+            let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
             let offsets: Vec<i64> = json_col
                 .offset
                 .unwrap()
@@ -711,8 +709,7 @@ pub fn array_from_json(
         }
         DataType::FixedSizeList(child_field, _) => {
             let children = json_col.children.clone().unwrap();
-            let child_array =
-                array_from_json(child_field, children.get(0).unwrap().clone(), dictionaries)?;
+            let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
             let null_buf = create_null_buf(&json_col);
             let list_data = ArrayData::builder(field.data_type().clone())
                 .len(json_col.count)
@@ -813,8 +810,7 @@ pub fn array_from_json(
         DataType::Map(child_field, _) => {
             let null_buf = create_null_buf(&json_col);
             let children = json_col.children.clone().unwrap();
-            let child_array =
-                array_from_json(child_field, children.get(0).unwrap().clone(), dictionaries)?;
+            let child_array = array_from_json(child_field, children[0].clone(), dictionaries)?;
             let offsets: Vec<i32> = json_col
                 .offset
                 .unwrap()
diff --git a/arrow-integration-testing/src/lib.rs b/arrow-integration-testing/src/lib.rs
index 553e69b0a..4ce7b06a1 100644
--- a/arrow-integration-testing/src/lib.rs
+++ b/arrow-integration-testing/src/lib.rs
@@ -78,10 +78,10 @@ pub fn canonicalize_schema(schema: &Schema) -> Schema {
         .map(|field| match field.data_type() {
             DataType::Map(child_field, sorted) => match child_field.data_type() {
                 DataType::Struct(fields) if fields.len() == 2 => {
-                    let first_field = fields.get(0).unwrap();
+                    let first_field = &fields[0];
                     let key_field =
                         Arc::new(Field::new("key", first_field.data_type().clone(), false));
-                    let second_field = fields.get(1).unwrap();
+                    let second_field = &fields[1];
                     let value_field = Arc::new(Field::new(
                         "value",
                         second_field.data_type().clone(),