[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2509: Replace azure sdk with custom implementation

[GitBox <gi...@apache.org>]

tustvold commented on code in PR #2509:
URL: https://github.com/apache/arrow-rs/pull/2509#discussion_r950711889


##########
object_store/src/azure/client.rs:
##########
@@ -0,0 +1,740 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::credential::{AzureCredential, CredentialProvider};
+use crate::azure::credential::*;
+use crate::client::pagination::stream_paginated;
+use crate::client::retry::RetryExt;
+use crate::path::DELIMITER;
+use crate::util::{encode_path, format_http_range, format_prefix};
+use crate::{BoxStream, ListResult, ObjectMeta, Path, Result, RetryConfig, StreamExt};
+use bytes::{Buf, Bytes};
+use chrono::{DateTime, TimeZone, Utc};
+use itertools::Itertools;
+use reqwest::{
+    header::{HeaderValue, CONTENT_LENGTH, IF_NONE_MATCH},
+    Client as ReqwestClient, Method, Response, StatusCode,
+};
+use serde::{Deserialize, Deserializer, Serialize};
+use snafu::{ResultExt, Snafu};
+use std::collections::HashMap;
+use std::ops::Range;
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+pub(crate) enum Error {
+    #[snafu(display("Error performing get request {}: {}", path, source))]
+    GetRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing put request {}: {}", path, source))]
+    PutRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing delete request {}: {}", path, source))]
+    DeleteRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing copy request {}: {}", path, source))]
+    CopyRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing list request: {}", source))]
+    ListRequest { source: reqwest::Error },
+
+    #[snafu(display("Error performing create multipart request: {}", source))]
+    CreateMultipartRequest { source: reqwest::Error },
+
+    #[snafu(display("Error performing complete multipart request: {}", source))]
+    CompleteMultipartRequest { source: reqwest::Error },
+
+    #[snafu(display("Got invalid list response: {}", source))]
+    InvalidListResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Got invalid multipart response: {}", source))]
+    InvalidMultipartResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Error authorizing request: {}", source))]
+    Authorization { source: crate::client::oauth::Error },
+}
+
+impl From<Error> for crate::Error {
+    fn from(err: Error) -> Self {
+        match err {
+            Error::GetRequest { source, path }
+            | Error::DeleteRequest { source, path }
+            | Error::CopyRequest { source, path }
+            | Error::PutRequest { source, path }
+                if matches!(source.status(), Some(StatusCode::NOT_FOUND)) =>
+            {
+                Self::NotFound {
+                    path,
+                    source: Box::new(source),
+                }
+            }
+            Error::CopyRequest { source, path }
+                if matches!(source.status(), Some(StatusCode::CONFLICT)) =>
+            {
+                Self::AlreadyExists {
+                    path,
+                    source: Box::new(source),
+                }
+            }
+            _ => Self::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(err),
+            },
+        }
+    }
+}
+
+/// Configuration for [AzureClient]
+#[derive(Debug)]
+pub struct AzureConfig {
+    pub account: String,
+    pub container: String,
+    pub credentials: CredentialProvider,
+    pub retry_config: RetryConfig,
+    pub allow_http: bool,
+    pub service: String,
+    pub is_emulator: bool,
+}
+
+impl AzureConfig {
+    fn path_url(&self, path: &Path) -> String {
+        if self.is_emulator {
+            format!(
+                "{}/{}/{}/{}",
+                self.service,
+                self.account,
+                self.container,
+                encode_path(path)
+            )
+        } else {
+            format!("{}/{}/{}", self.service, self.container, encode_path(path))
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct AzureClient {
+    config: AzureConfig,
+    client: ReqwestClient,
+}
+
+impl AzureClient {
+    /// create a new instance of [AzureClient]
+    pub fn new(config: AzureConfig) -> Self {
+        let client = reqwest::ClientBuilder::new()
+            .https_only(!config.allow_http)
+            .build()
+            .unwrap();
+
+        Self { config, client }
+    }
+
+    /// Returns the config
+    pub fn config(&self) -> &AzureConfig {
+        &self.config
+    }
+
+    async fn get_credential(&self) -> Result<AzureCredential> {
+        match &self.config.credentials {
+            CredentialProvider::AccessKey(key) => {
+                Ok(AzureCredential::AccessKey(key.to_owned()))
+            }
+            CredentialProvider::ClientSecret(cred) => {
+                let token = cred
+                    .fetch_token(&self.client, &self.config.retry_config)
+                    .await
+                    .context(AuthorizationSnafu)?;
+                Ok(AzureCredential::AuthorizationToken(
+                    // we do the conversion to a HeaderValue here, since it is fallible
+                    // and we wna to use it in an infallible function
+                    HeaderValue::from_str(&format!("Bearer {}", token)).map_err(
+                        |err| crate::Error::Generic {
+                            store: "MicrosoftAzure",
+                            source: Box::new(err),
+                        },
+                    )?,
+                ))
+            }
+            CredentialProvider::SASToken(sas) => {
+                Ok(AzureCredential::SASToken(sas.clone()))
+            }
+        }
+    }
+
+    /// Make an Azure PUT request <https://docs.microsoft.com/en-us/rest/api/storageservices/put-blob>
+    pub async fn put_request<T: Serialize + ?Sized + Sync>(
+        &self,
+        path: &Path,
+        bytes: Option<Bytes>,
+        is_block_op: bool,
+        query: &T,
+    ) -> Result<Response> {
+        let credential = self.get_credential().await?;
+        let url = self.config.path_url(path);
+
+        let mut builder = self.client.request(Method::PUT, url);
+
+        if !is_block_op {
+            builder = builder.header(&BLOB_TYPE, "BlockBlob");
+        }
+
+        if let Some(bytes) = bytes {
+            builder = builder
+                .header(CONTENT_LENGTH, HeaderValue::from(bytes.len()))
+                .body(bytes)
+        }
+
+        let response = builder
+            .query(query)
+            .with_azure_authorization(&credential, &self.config.account)
+            .send_retry(&self.config.retry_config)
+            .await
+            .context(PutRequestSnafu {
+                path: path.as_ref(),
+            })?
+            .error_for_status()
+            .context(PutRequestSnafu {
+                path: path.as_ref(),
+            })?;
+
+        Ok(response)
+    }
+
+    /// Make an Azure GET request
+    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob>
+    /// <https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties>
+    pub async fn get_request(
+        &self,
+        path: &Path,
+        range: Option<Range<usize>>,
+        head: bool,
+    ) -> Result<Response> {
+        let credential = self.get_credential().await?;
+        let url = self.config.path_url(path);
+        let method = match head {
+            true => Method::HEAD,
+            false => Method::GET,
+        };
+
+        let mut builder = self
+            .client
+            .request(method, url)
+            .header(CONTENT_LENGTH, HeaderValue::from_static("0"))
+            .body(Bytes::new());
+
+        if let Some(range) = range {
+            // Note: Azurite emulator does not support crc64 headers
+            if !self.config.is_emulator && range.end - range.start < 1024 * 1024 * 4 {
+                builder = builder
+                    .header(&RANGE_GET_CONTENT_CRC64, HeaderValue::from_static("true"));

Review Comment:
   Do we need to specify this, AFAICT we never consume its output?



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys

Review Comment:
   ```suggestion
   /// Generate signed key for authorization via access keys
   /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key>
   ```



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";

Review Comment:
   I think RFC2822 is equivalent and has native support in chrono



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {

Review Comment:
   ```suggestion
   /// <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 {
   ```



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h

Review Comment:
   FYI there is a slightly lighter weight version of this - https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#blob-queue-and-file-services-shared-key-lite-authorization



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");

Review Comment:
   I think you can just use a normal Range header, this only exists for the case where using an HTTP library that doesn't let you override it



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))

Review Comment:
   ```suggestion
           .flatten()
   ```
   ?



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {

Review Comment:
   ```suggestion
   /// <https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#constructing-the-canonicalized-headers-string>
   fn canonicalize_header(headers: &HeaderMap) -> String {
   ```



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");
+    }
+    result
+}
+
+fn canonicalized_resource(account: &str, uri: &Url) -> String {
+    let mut can_res: String = String::new();
+    can_res += "/";
+    can_res += account;
+
+    for p in uri.path_segments().into_iter().flatten() {
+        can_res.push('/');
+        can_res.push_str(p);
+    }

Review Comment:
   ```suggestion
       let mut can_res = uri.path().to_string()
   ```



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");
+    }
+    result
+}
+
+fn canonicalized_resource(account: &str, uri: &Url) -> String {
+    let mut can_res: String = String::new();
+    can_res += "/";
+    can_res += account;
+
+    for p in uri.path_segments().into_iter().flatten() {
+        can_res.push('/');
+        can_res.push_str(p);
+    }
+    can_res += "\n";

Review Comment:
   ```suggestion
       can_res.push('\n');
   ```
   
   Not actually sure if there is a difference



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");
+    }
+    result
+}
+
+fn canonicalized_resource(account: &str, uri: &Url) -> String {

Review Comment:
   ```suggestion
   /// <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 {
   ```



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");
+    }
+    result
+}
+
+fn canonicalized_resource(account: &str, uri: &Url) -> String {
+    let mut can_res: String = String::new();
+    can_res += "/";
+    can_res += account;
+
+    for p in uri.path_segments().into_iter().flatten() {
+        can_res.push('/');
+        can_res.push_str(p);
+    }
+    can_res += "\n";
+
+    // query parameters

Review Comment:
   FWIW the AWS version I wrote just requires the client to provide query arguments in lexicographic order



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);

Review Comment:
   This is like they're taunting us, "we know how to do request authorization without an obnoxious song and dance", but we'll only do it sometimes :rage: 



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");

Review Comment:
   It might be marginally faster to use `push_str`, etc... instead of `format!`



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");
+    }
+    result
+}
+
+fn canonicalized_resource(account: &str, uri: &Url) -> String {
+    let mut can_res: String = String::new();
+    can_res += "/";
+    can_res += account;
+
+    for p in uri.path_segments().into_iter().flatten() {
+        can_res.push('/');
+        can_res.push_str(p);
+    }
+    can_res += "\n";
+
+    // query parameters
+    let query_pairs = uri.query_pairs();
+    {
+        let mut qps: Vec<String> = Vec::new();

Review Comment:
   I think using a `BTreeMap<&str, Vec<&str>>` would avoid some copies, be faster, and easier to follow.
   
   Something like (not tested)
   
   ```
   let mut qps: BTreeMap<&str, Vec<&str>> = Default::default();
   
   for (q, v) in uri.query_pairs() {
       qps.entry(q.as_ref()).or_default().push(v.as_ref())
   }
   ```
   
   You could then just iterate the `BTreeMap` to produce the output
   
   ```
   for (key, values) in qps {
       values.sort_unstable();
       ...
   }
   ```



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)

Review Comment:
   You could use the standard `Date` header here



##########
object_store/src/azure/credential.rs:
##########
@@ -0,0 +1,254 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::client::oauth::ClientSecretOAuthProvider;
+use crate::util::hmac_sha256;
+use chrono::Utc;
+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,
+    },
+    Method, RequestBuilder,
+};
+use std::borrow::Cow;
+use std::str;
+use url::Url;
+
+static AZURE_VERSION: HeaderValue = HeaderValue::from_static("2021-08-06");
+static VERSION: HeaderName = HeaderName::from_static("x-ms-version");
+pub(crate) static RANGE_GET_CONTENT_CRC64: HeaderName =
+    HeaderName::from_static("x-ms-range-get-content-crc64");
+pub(crate) static MS_RANGE: HeaderName = HeaderName::from_static("x-ms-range");
+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");
+pub(crate) static COPY_SOURCE: HeaderName = HeaderName::from_static("x-ms-copy-source");
+static CONTENT_MD5: HeaderName = HeaderName::from_static("content-md5");
+pub(crate) static RFC1123_FMT: &str = "%a, %d %h %Y %T GMT";
+
+/// Provides credentials for use when signing requests
+#[derive(Debug)]
+pub enum CredentialProvider {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    ClientSecret(ClientSecretOAuthProvider),
+}
+
+pub(crate) enum AzureCredential {
+    AccessKey(String),
+    SASToken(Vec<(String, String)>),
+    AuthorizationToken(HeaderValue),
+}
+
+/// A list of known Azure authority hosts
+pub mod authority_hosts {
+    /// China-based Azure Authority Host
+    pub const AZURE_CHINA: &str = "https://login.chinacloudapi.cn";
+    /// Germany-based Azure Authority Host
+    pub const AZURE_GERMANY: &str = "https://login.microsoftonline.de";
+    /// US Government Azure Authority Host
+    pub const AZURE_GOVERNMENT: &str = "https://login.microsoftonline.us";
+    /// Public Cloud Azure Authority Host
+    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;
+}
+
+impl CredentialExt for RequestBuilder {
+    fn with_azure_authorization(
+        mut self,
+        credential: &AzureCredential,
+        account: &str,
+    ) -> Self {
+        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("x-ms-date", &date_val)
+            .header(&VERSION, &AZURE_VERSION);
+
+        // Hack around lack of access to underlying request
+        // https://github.com/seanmonstar/reqwest/issues/1212
+        let request = self
+            .try_clone()
+            .expect("not stream")
+            .build()
+            .expect("request valid");
+
+        match credential {
+            AzureCredential::AccessKey(key) => {
+                let signature = generate_authorization(
+                    request.headers(),
+                    request.url(),
+                    request.method(),
+                    account,
+                    key.as_str(),
+                );
+                self = self
+                    // "signature" is a base 64 encoded string so it should never contain illegal characters.
+                    .header(
+                        AUTHORIZATION,
+                        HeaderValue::from_str(signature.as_str()).unwrap(),
+                    );
+            }
+            AzureCredential::AuthorizationToken(token) => {
+                self = self.header(AUTHORIZATION, token);
+            }
+            AzureCredential::SASToken(query_pairs) => {
+                self = self.query(&query_pairs);
+            }
+        };
+
+        self
+    }
+}
+
+// generate signed key for authorization via access keys
+fn generate_authorization(
+    h: &HeaderMap,
+    u: &Url,
+    method: &Method,
+    account: &str,
+    key: &str,
+) -> String {
+    let str_to_sign = string_to_sign(h, u, method, account);
+    let auth = hmac_sha256(base64::decode(key).unwrap(), &str_to_sign);
+    format!("SharedKey {}:{}", account, base64::encode(auth))
+}
+
+fn add_if_exists<'a>(h: &'a HeaderMap, key: &HeaderName) -> &'a str {
+    h.get(key)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .unwrap_or_default()
+}
+
+fn string_to_sign(h: &HeaderMap, u: &Url, method: &Method, account: &str) -> String {
+    // content length must only be specified if != 0
+    // this is valid from 2015-02-21
+    let content_length = h
+        .get(&CONTENT_LENGTH)
+        .map(|s| s.to_str())
+        .transpose()
+        .ok()
+        .unwrap_or(Some(""))
+        .filter(|&v| v != "0")
+        .unwrap_or_default();
+    format!(
+        "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}{}",
+        method.as_ref(),
+        add_if_exists(h, &CONTENT_ENCODING),
+        add_if_exists(h, &CONTENT_LANGUAGE),
+        content_length,
+        add_if_exists(h, &CONTENT_MD5),
+        add_if_exists(h, &CONTENT_TYPE),
+        add_if_exists(h, &DATE),
+        add_if_exists(h, &IF_MODIFIED_SINCE),
+        add_if_exists(h, &IF_MATCH),
+        add_if_exists(h, &IF_NONE_MATCH),
+        add_if_exists(h, &IF_UNMODIFIED_SINCE),
+        add_if_exists(h, &RANGE),
+        canonicalize_header(h),
+        canonicalized_resource(account, u)
+    )
+}
+
+fn canonicalize_header(headers: &HeaderMap) -> String {
+    let mut names = headers
+        .iter()
+        .filter_map(|(k, _)| {
+            (k.as_str().starts_with("x-ms"))
+                // TODO remove unwraps
+                .then(|| (k.as_str(), headers.get(k).unwrap().to_str().unwrap()))
+        })
+        .collect::<Vec<_>>();
+    names.sort_unstable();
+
+    let mut result = String::new();
+    for (name, value) in names {
+        result = format!("{result}{name}:{value}\n");
+    }
+    result
+}
+
+fn canonicalized_resource(account: &str, uri: &Url) -> String {

Review Comment:
   FYI again there is a simpler API for this https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key#shared-key-lite-and-table-service-format-for-2009-09-19-and-later



##########
object_store/src/azure/mod.rs:
##########
@@ -0,0 +1,711 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for Azure blob storage
+//!
+//! ## Streaming uploads
+//!
+//! [ObjectStore::put_multipart] will upload data in blocks and write a blob from those
+//! blocks. Data is buffered internally to make blocks of at least 5MB and blocks
+//! are uploaded concurrently.
+//!
+//! [ObjectStore::abort_multipart] is a no-op, since Azure Blob Store doesn't provide
+//! a way to drop old blocks. Instead unused blocks are automatically cleaned up
+//! after 7 days.
+use self::client::{BlockId, BlockList};
+use crate::{
+    multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart},
+    path::Path,
+    GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result, RetryConfig,
+};
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{TimeZone, Utc};
+use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use snafu::{ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use url::Url;
+
+pub use credential::authority_hosts;
+
+mod client;
+mod credential;
+
+/// The well-known account used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT: &str = "devstoreaccount1";
+
+/// The well-known account key used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT_KEY: &str =
+    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+
+/// A specialized `Error` for Azure object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+    #[snafu(display("Last-Modified Header missing from response"))]
+    MissingLastModified,
+
+    #[snafu(display("Content-Length Header missing from response"))]
+    MissingContentLength,
+
+    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+    InvalidLastModified {
+        last_modified: String,
+        source: chrono::ParseError,
+    },
+
+    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+    InvalidContentLength {
+        content_length: String,
+        source: std::num::ParseIntError,
+    },
+
+    #[snafu(display("Received header containing non-ASCII data"))]
+    BadHeader { source: reqwest::header::ToStrError },
+
+    #[snafu(display("Unable parse source url. Url: {}, Error: {}", url, source))]
+    UnableToParseUrl {
+        source: url::ParseError,
+        url: String,
+    },
+
+    #[snafu(display(
+        "Unable parse emulator url {}={}, Error: {}",
+        env_name,
+        env_value,
+        source
+    ))]
+    UnableToParseEmulatorUrl {
+        env_name: String,
+        env_value: String,
+        source: url::ParseError,
+    },
+
+    #[snafu(display("Account must be specified"))]
+    MissingAccount {},
+
+    #[snafu(display("Container name must be specified"))]
+    MissingContainerName {},
+
+    #[snafu(display("At least one authorization option must be specified"))]
+    MissingCredentials {},
+}
+
+impl From<Error> for super::Error {
+    fn from(source: Error) -> Self {
+        Self::Generic {
+            store: "MicrosoftAzure",
+            source: Box::new(source),
+        }
+    }
+}
+
+/// Interface for [Microsoft Azure Blob Storage](https://azure.microsoft.com/en-us/services/storage/blobs/).
+#[derive(Debug)]
+pub struct MicrosoftAzure {
+    client: Arc<client::AzureClient>,
+}
+
+impl std::fmt::Display for MicrosoftAzure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzure {{ account: {}, container: {} }}",
+            self.client.config().account,
+            self.client.config().container
+        )
+    }
+}
+
+#[async_trait]
+impl ObjectStore for MicrosoftAzure {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+        self.client
+            .put_request(location, Some(bytes), false, &())
+            .await?;
+        Ok(())
+    }
+
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        let inner = AzureMultiPartUpload {
+            client: Arc::clone(&self.client),
+            location: location.to_owned(),
+        };
+        Ok((String::new(), Box::new(CloudMultiPartUpload::new(inner, 8))))
+    }
+
+    async fn abort_multipart(
+        &self,
+        _location: &Path,
+        _multipart_id: &MultipartId,
+    ) -> Result<()> {
+        // There is no way to drop blocks that have been uploaded. Instead, they simply
+        // expire in 7 days.
+        Ok(())
+    }
+
+    async fn get(&self, location: &Path) -> Result<GetResult> {
+        let response = self.client.get_request(location, None, false).await?;
+        let stream = response
+            .bytes_stream()
+            .map_err(|source| crate::Error::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(source),
+            })
+            .boxed();
+
+        Ok(GetResult::Stream(stream))
+    }
+
+    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+        let bytes = self
+            .client
+            .get_request(location, Some(range), false)
+            .await?
+            .bytes()
+            .await
+            .map_err(|source| client::Error::GetRequest {
+                source,
+                path: location.to_string(),
+            })?;
+        Ok(bytes)
+    }
+
+    async fn head(&self, location: &Path) -> Result<ObjectMeta> {
+        use reqwest::header::{CONTENT_LENGTH, LAST_MODIFIED};
+
+        // Extract meta from headers
+        // https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties
+        let response = self.client.get_request(location, None, true).await?;
+        let headers = response.headers();
+
+        let last_modified = headers
+            .get(LAST_MODIFIED)
+            .ok_or(Error::MissingLastModified)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let last_modified = Utc
+            .datetime_from_str(last_modified, credential::RFC1123_FMT)
+            .context(InvalidLastModifiedSnafu { last_modified })?;
+
+        let content_length = headers
+            .get(CONTENT_LENGTH)
+            .ok_or(Error::MissingContentLength)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let content_length = content_length
+            .parse()
+            .context(InvalidContentLengthSnafu { content_length })?;
+
+        Ok(ObjectMeta {
+            location: location.clone(),
+            last_modified,
+            size: content_length,
+        })
+    }
+
+    async fn delete(&self, location: &Path) -> Result<()> {
+        self.client.delete_request(location, &()).await
+    }
+
+    async fn list(
+        &self,
+        prefix: Option<&Path>,
+    ) -> Result<BoxStream<'_, Result<ObjectMeta>>> {
+        let stream = self
+            .client
+            .list_paginated(prefix, false)
+            .map_ok(|r| futures::stream::iter(r.objects.into_iter().map(Ok)))
+            .try_flatten()
+            .boxed();
+
+        Ok(stream)
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        let mut stream = self.client.list_paginated(prefix, true);
+
+        let mut common_prefixes = BTreeSet::new();
+        let mut objects = Vec::new();
+
+        while let Some(result) = stream.next().await {
+            let response = result?;
+            common_prefixes.extend(response.common_prefixes.into_iter());
+            objects.extend(response.objects.into_iter());
+        }
+
+        Ok(ListResult {
+            common_prefixes: common_prefixes.into_iter().collect(),
+            objects,
+        })
+    }
+
+    async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, true).await
+    }
+
+    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, false).await
+    }
+}
+
+// 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
+// complete -> PUT block list
+// abort -> No equivalent; blocks are simply dropped after 7 days

Review Comment:
   ```suggestion
   /// 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
   /// complete -> PUT block list
   /// abort -> No equivalent; blocks are simply dropped after 7 days
   ```



##########
object_store/src/client/oauth.rs:
##########
@@ -219,3 +223,82 @@ fn b64_encode_obj<T: serde::Serialize>(obj: &T) -> Result<String> {
     let string = serde_json::to_string(obj).context(EncodeSnafu)?;
     Ok(base64::encode_config(string, base64::URL_SAFE_NO_PAD))
 }
+

Review Comment:
   There might be unused code warnings when just GCP or just Azure is enabled, not sure. Might need to split the module or something



##########
object_store/src/azure/mod.rs:
##########
@@ -0,0 +1,711 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for Azure blob storage
+//!
+//! ## Streaming uploads
+//!
+//! [ObjectStore::put_multipart] will upload data in blocks and write a blob from those
+//! blocks. Data is buffered internally to make blocks of at least 5MB and blocks
+//! are uploaded concurrently.
+//!
+//! [ObjectStore::abort_multipart] is a no-op, since Azure Blob Store doesn't provide
+//! a way to drop old blocks. Instead unused blocks are automatically cleaned up
+//! after 7 days.
+use self::client::{BlockId, BlockList};
+use crate::{
+    multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart},
+    path::Path,
+    GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result, RetryConfig,
+};
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{TimeZone, Utc};
+use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use snafu::{ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use url::Url;
+
+pub use credential::authority_hosts;
+
+mod client;
+mod credential;
+
+/// The well-known account used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT: &str = "devstoreaccount1";
+
+/// The well-known account key used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT_KEY: &str =
+    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+
+/// A specialized `Error` for Azure object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+    #[snafu(display("Last-Modified Header missing from response"))]
+    MissingLastModified,
+
+    #[snafu(display("Content-Length Header missing from response"))]
+    MissingContentLength,
+
+    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+    InvalidLastModified {
+        last_modified: String,
+        source: chrono::ParseError,
+    },
+
+    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+    InvalidContentLength {
+        content_length: String,
+        source: std::num::ParseIntError,
+    },
+
+    #[snafu(display("Received header containing non-ASCII data"))]
+    BadHeader { source: reqwest::header::ToStrError },
+
+    #[snafu(display("Unable parse source url. Url: {}, Error: {}", url, source))]
+    UnableToParseUrl {
+        source: url::ParseError,
+        url: String,
+    },
+
+    #[snafu(display(
+        "Unable parse emulator url {}={}, Error: {}",
+        env_name,
+        env_value,
+        source
+    ))]
+    UnableToParseEmulatorUrl {
+        env_name: String,
+        env_value: String,
+        source: url::ParseError,
+    },
+
+    #[snafu(display("Account must be specified"))]
+    MissingAccount {},
+
+    #[snafu(display("Container name must be specified"))]
+    MissingContainerName {},
+
+    #[snafu(display("At least one authorization option must be specified"))]
+    MissingCredentials {},
+}
+
+impl From<Error> for super::Error {
+    fn from(source: Error) -> Self {
+        Self::Generic {
+            store: "MicrosoftAzure",
+            source: Box::new(source),
+        }
+    }
+}
+
+/// Interface for [Microsoft Azure Blob Storage](https://azure.microsoft.com/en-us/services/storage/blobs/).
+#[derive(Debug)]
+pub struct MicrosoftAzure {
+    client: Arc<client::AzureClient>,
+}
+
+impl std::fmt::Display for MicrosoftAzure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzure {{ account: {}, container: {} }}",
+            self.client.config().account,
+            self.client.config().container
+        )
+    }
+}
+
+#[async_trait]
+impl ObjectStore for MicrosoftAzure {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+        self.client
+            .put_request(location, Some(bytes), false, &())
+            .await?;
+        Ok(())
+    }
+
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        let inner = AzureMultiPartUpload {
+            client: Arc::clone(&self.client),
+            location: location.to_owned(),
+        };
+        Ok((String::new(), Box::new(CloudMultiPartUpload::new(inner, 8))))
+    }
+
+    async fn abort_multipart(
+        &self,
+        _location: &Path,
+        _multipart_id: &MultipartId,
+    ) -> Result<()> {
+        // There is no way to drop blocks that have been uploaded. Instead, they simply
+        // expire in 7 days.
+        Ok(())
+    }
+
+    async fn get(&self, location: &Path) -> Result<GetResult> {
+        let response = self.client.get_request(location, None, false).await?;
+        let stream = response
+            .bytes_stream()
+            .map_err(|source| crate::Error::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(source),
+            })
+            .boxed();
+
+        Ok(GetResult::Stream(stream))
+    }
+
+    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+        let bytes = self
+            .client
+            .get_request(location, Some(range), false)
+            .await?
+            .bytes()
+            .await
+            .map_err(|source| client::Error::GetRequest {
+                source,
+                path: location.to_string(),
+            })?;
+        Ok(bytes)
+    }
+
+    async fn head(&self, location: &Path) -> Result<ObjectMeta> {
+        use reqwest::header::{CONTENT_LENGTH, LAST_MODIFIED};
+
+        // Extract meta from headers
+        // https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties
+        let response = self.client.get_request(location, None, true).await?;
+        let headers = response.headers();
+
+        let last_modified = headers
+            .get(LAST_MODIFIED)
+            .ok_or(Error::MissingLastModified)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let last_modified = Utc
+            .datetime_from_str(last_modified, credential::RFC1123_FMT)
+            .context(InvalidLastModifiedSnafu { last_modified })?;
+
+        let content_length = headers
+            .get(CONTENT_LENGTH)
+            .ok_or(Error::MissingContentLength)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let content_length = content_length
+            .parse()
+            .context(InvalidContentLengthSnafu { content_length })?;
+
+        Ok(ObjectMeta {
+            location: location.clone(),
+            last_modified,
+            size: content_length,
+        })
+    }
+
+    async fn delete(&self, location: &Path) -> Result<()> {
+        self.client.delete_request(location, &()).await
+    }
+
+    async fn list(
+        &self,
+        prefix: Option<&Path>,
+    ) -> Result<BoxStream<'_, Result<ObjectMeta>>> {
+        let stream = self
+            .client
+            .list_paginated(prefix, false)
+            .map_ok(|r| futures::stream::iter(r.objects.into_iter().map(Ok)))
+            .try_flatten()
+            .boxed();
+
+        Ok(stream)
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        let mut stream = self.client.list_paginated(prefix, true);
+
+        let mut common_prefixes = BTreeSet::new();
+        let mut objects = Vec::new();
+
+        while let Some(result) = stream.next().await {
+            let response = result?;
+            common_prefixes.extend(response.common_prefixes.into_iter());
+            objects.extend(response.objects.into_iter());
+        }
+
+        Ok(ListResult {
+            common_prefixes: common_prefixes.into_iter().collect(),
+            objects,
+        })
+    }
+
+    async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, true).await
+    }
+
+    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, false).await
+    }
+}
+
+// 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
+// complete -> PUT block list
+// abort -> No equivalent; blocks are simply dropped after 7 days
+#[derive(Debug, Clone)]
+struct AzureMultiPartUpload {
+    client: Arc<client::AzureClient>,
+    location: Path,
+}
+
+#[async_trait]
+impl CloudMultiPartUploadImpl for AzureMultiPartUpload {
+    async fn put_multipart_part(
+        &self,
+        buf: Vec<u8>,
+        part_idx: usize,
+    ) -> Result<UploadPart, io::Error> {
+        let content_id = format!("{:20}", part_idx);
+        let block_id: BlockId = content_id.clone().into();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(buf.into()),
+                true,
+                &[("comp", "block"), ("blockid", &base64::encode(block_id))],
+            )
+            .await?;
+
+        Ok(UploadPart { content_id })
+    }
+
+    async fn complete(&self, completed_parts: Vec<UploadPart>) -> Result<(), io::Error> {
+        let blocks = completed_parts
+            .into_iter()
+            .map(|part| BlockId::from(part.content_id))
+            .collect();
+
+        let block_list = BlockList { blocks };
+        let block_xml = block_list.to_xml();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(block_xml.into()),
+                true,
+                &[("comp", "blocklist")],
+            )
+            .await?;
+
+        Ok(())
+    }
+}
+
+/// Configure a connection to Microsoft Azure Blob Storage container using
+/// the specified credentials.
+///
+/// # Example
+/// ```
+/// # let ACCOUNT = "foo";
+/// # let BUCKET_NAME = "foo";
+/// # let ACCESS_KEY = "foo";
+/// # use object_store::azure::MicrosoftAzureBuilder;
+/// let azure = MicrosoftAzureBuilder::new()
+///  .with_account_name(ACCOUNT)
+///  .with_access_key_authorization(ACCESS_KEY)
+///  .with_container_name(BUCKET_NAME)
+///  .build();
+/// ```
+#[derive(Default)]
+pub struct MicrosoftAzureBuilder {
+    account_name: Option<String>,
+    access_key: Option<String>,
+    container_name: Option<String>,
+    bearer_token: Option<String>,
+    client_id: Option<String>,
+    client_secret: Option<String>,
+    tenant_id: Option<String>,
+    sas_query_pairs: Option<Vec<(String, String)>>,
+    authority_host: Option<String>,
+    use_emulator: bool,
+    retry_config: RetryConfig,
+    allow_http: bool,
+}
+
+impl Debug for MicrosoftAzureBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzureBuilder {{ account: {:?}, container_name: {:?} }}",
+            self.account_name, self.container_name
+        )
+    }
+}
+
+impl MicrosoftAzureBuilder {
+    /// Create a new [`MicrosoftAzureBuilder`] with default values.
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    /// Create an instance of [MicrosoftAzureBuilder] with values pre-populated from environment variables.
+    ///
+    /// Variables extracted from environment:
+    /// * AZURE_STORAGE_ACCOUNT_NAME: storage account name
+    /// * AZURE_STORAGE_ACCOUNT_KEY: storage account master key
+    /// * AZURE_STORAGE_ACCESS_KEY: alias for AZURE_STORAGE_ACCOUNT_KEY
+    /// * AZURE_STORAGE_CLIENT_ID -> client id for service principal authorization
+    /// * AZURE_STORAGE_CLIENT_SECRET -> client secret for service principal authorization
+    /// * AZURE_STORAGE_TENANT_ID -> tenant id used in oauth flows
+    /// # Example
+    /// ```
+    /// use object_store::azure::MicrosoftAzureBuilder;
+    ///
+    /// let azure = MicrosoftAzureBuilder::from_env()
+    ///     .with_container_name("foo")
+    ///     .build();
+    /// ```
+    pub fn from_env() -> Self {
+        let mut builder = Self::default();
+
+        if let Ok(account_name) = std::env::var("AZURE_STORAGE_ACCOUNT_NAME") {
+            builder.account_name = Some(account_name);
+        }
+
+        if let Ok(access_key) = std::env::var("AZURE_STORAGE_ACCOUNT_KEY") {
+            builder.access_key = Some(access_key);
+        } else if let Ok(access_key) = std::env::var("AZURE_STORAGE_ACCESS_KEY") {
+            builder.access_key = Some(access_key);
+        }
+
+        if let Ok(client_id) = std::env::var("AZURE_STORAGE_CLIENT_ID") {
+            builder.client_id = Some(client_id);
+        }
+
+        if let Ok(client_secret) = std::env::var("AZURE_STORAGE_CLIENT_SECRET") {
+            builder.client_secret = Some(client_secret);
+        }
+
+        if let Ok(tenant_id) = std::env::var("AZURE_STORAGE_TENANT_ID") {
+            builder.tenant_id = Some(tenant_id);
+        }
+
+        builder
+    }
+
+    /// Set the Azure Account (required)
+    pub fn with_account_name(mut self, account: impl Into<String>) -> Self {
+        self.account_name = Some(account.into());
+        self
+    }
+
+    /// Set the Azure Container Name (required)
+    pub fn with_container_name(mut self, container_name: impl Into<String>) -> Self {
+        self.container_name = Some(container_name.into());
+        self
+    }
+
+    /// Set the Azure Access Key (required - one of access key, bearer token, or client credentials)
+    pub fn with_access_key_authorization(

Review Comment:
   The _authorization suffix has been added



##########
object_store/src/azure/mod.rs:
##########
@@ -0,0 +1,711 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for Azure blob storage
+//!
+//! ## Streaming uploads
+//!
+//! [ObjectStore::put_multipart] will upload data in blocks and write a blob from those
+//! blocks. Data is buffered internally to make blocks of at least 5MB and blocks
+//! are uploaded concurrently.
+//!
+//! [ObjectStore::abort_multipart] is a no-op, since Azure Blob Store doesn't provide
+//! a way to drop old blocks. Instead unused blocks are automatically cleaned up
+//! after 7 days.
+use self::client::{BlockId, BlockList};
+use crate::{
+    multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart},
+    path::Path,
+    GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result, RetryConfig,
+};
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{TimeZone, Utc};
+use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use snafu::{ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use url::Url;
+
+pub use credential::authority_hosts;
+
+mod client;
+mod credential;
+
+/// The well-known account used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT: &str = "devstoreaccount1";
+
+/// The well-known account key used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT_KEY: &str =
+    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+
+/// A specialized `Error` for Azure object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+    #[snafu(display("Last-Modified Header missing from response"))]
+    MissingLastModified,
+
+    #[snafu(display("Content-Length Header missing from response"))]
+    MissingContentLength,
+
+    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+    InvalidLastModified {
+        last_modified: String,
+        source: chrono::ParseError,
+    },
+
+    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+    InvalidContentLength {
+        content_length: String,
+        source: std::num::ParseIntError,
+    },
+
+    #[snafu(display("Received header containing non-ASCII data"))]
+    BadHeader { source: reqwest::header::ToStrError },
+
+    #[snafu(display("Unable parse source url. Url: {}, Error: {}", url, source))]
+    UnableToParseUrl {
+        source: url::ParseError,
+        url: String,
+    },
+
+    #[snafu(display(
+        "Unable parse emulator url {}={}, Error: {}",
+        env_name,
+        env_value,
+        source
+    ))]
+    UnableToParseEmulatorUrl {
+        env_name: String,
+        env_value: String,
+        source: url::ParseError,
+    },
+
+    #[snafu(display("Account must be specified"))]
+    MissingAccount {},
+
+    #[snafu(display("Container name must be specified"))]
+    MissingContainerName {},
+
+    #[snafu(display("At least one authorization option must be specified"))]
+    MissingCredentials {},
+}
+
+impl From<Error> for super::Error {
+    fn from(source: Error) -> Self {
+        Self::Generic {
+            store: "MicrosoftAzure",
+            source: Box::new(source),
+        }
+    }
+}
+
+/// Interface for [Microsoft Azure Blob Storage](https://azure.microsoft.com/en-us/services/storage/blobs/).
+#[derive(Debug)]
+pub struct MicrosoftAzure {
+    client: Arc<client::AzureClient>,
+}
+
+impl std::fmt::Display for MicrosoftAzure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzure {{ account: {}, container: {} }}",
+            self.client.config().account,
+            self.client.config().container
+        )
+    }
+}
+
+#[async_trait]
+impl ObjectStore for MicrosoftAzure {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+        self.client
+            .put_request(location, Some(bytes), false, &())
+            .await?;
+        Ok(())
+    }
+
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        let inner = AzureMultiPartUpload {
+            client: Arc::clone(&self.client),
+            location: location.to_owned(),
+        };
+        Ok((String::new(), Box::new(CloudMultiPartUpload::new(inner, 8))))
+    }
+
+    async fn abort_multipart(
+        &self,
+        _location: &Path,
+        _multipart_id: &MultipartId,
+    ) -> Result<()> {
+        // There is no way to drop blocks that have been uploaded. Instead, they simply
+        // expire in 7 days.
+        Ok(())
+    }
+
+    async fn get(&self, location: &Path) -> Result<GetResult> {
+        let response = self.client.get_request(location, None, false).await?;
+        let stream = response
+            .bytes_stream()
+            .map_err(|source| crate::Error::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(source),
+            })
+            .boxed();
+
+        Ok(GetResult::Stream(stream))
+    }
+
+    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+        let bytes = self
+            .client
+            .get_request(location, Some(range), false)
+            .await?
+            .bytes()
+            .await
+            .map_err(|source| client::Error::GetRequest {
+                source,
+                path: location.to_string(),
+            })?;
+        Ok(bytes)
+    }
+
+    async fn head(&self, location: &Path) -> Result<ObjectMeta> {
+        use reqwest::header::{CONTENT_LENGTH, LAST_MODIFIED};
+
+        // Extract meta from headers
+        // https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties
+        let response = self.client.get_request(location, None, true).await?;
+        let headers = response.headers();
+
+        let last_modified = headers
+            .get(LAST_MODIFIED)
+            .ok_or(Error::MissingLastModified)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let last_modified = Utc
+            .datetime_from_str(last_modified, credential::RFC1123_FMT)
+            .context(InvalidLastModifiedSnafu { last_modified })?;
+
+        let content_length = headers
+            .get(CONTENT_LENGTH)
+            .ok_or(Error::MissingContentLength)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let content_length = content_length
+            .parse()
+            .context(InvalidContentLengthSnafu { content_length })?;
+
+        Ok(ObjectMeta {
+            location: location.clone(),
+            last_modified,
+            size: content_length,
+        })
+    }
+
+    async fn delete(&self, location: &Path) -> Result<()> {
+        self.client.delete_request(location, &()).await
+    }
+
+    async fn list(
+        &self,
+        prefix: Option<&Path>,
+    ) -> Result<BoxStream<'_, Result<ObjectMeta>>> {
+        let stream = self
+            .client
+            .list_paginated(prefix, false)
+            .map_ok(|r| futures::stream::iter(r.objects.into_iter().map(Ok)))
+            .try_flatten()
+            .boxed();
+
+        Ok(stream)
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        let mut stream = self.client.list_paginated(prefix, true);
+
+        let mut common_prefixes = BTreeSet::new();
+        let mut objects = Vec::new();
+
+        while let Some(result) = stream.next().await {
+            let response = result?;
+            common_prefixes.extend(response.common_prefixes.into_iter());
+            objects.extend(response.objects.into_iter());
+        }
+
+        Ok(ListResult {
+            common_prefixes: common_prefixes.into_iter().collect(),
+            objects,
+        })
+    }
+
+    async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, true).await
+    }
+
+    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, false).await
+    }
+}
+
+// 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
+// complete -> PUT block list
+// abort -> No equivalent; blocks are simply dropped after 7 days
+#[derive(Debug, Clone)]
+struct AzureMultiPartUpload {
+    client: Arc<client::AzureClient>,
+    location: Path,
+}
+
+#[async_trait]
+impl CloudMultiPartUploadImpl for AzureMultiPartUpload {
+    async fn put_multipart_part(
+        &self,
+        buf: Vec<u8>,
+        part_idx: usize,
+    ) -> Result<UploadPart, io::Error> {
+        let content_id = format!("{:20}", part_idx);
+        let block_id: BlockId = content_id.clone().into();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(buf.into()),
+                true,
+                &[("comp", "block"), ("blockid", &base64::encode(block_id))],
+            )
+            .await?;
+
+        Ok(UploadPart { content_id })
+    }
+
+    async fn complete(&self, completed_parts: Vec<UploadPart>) -> Result<(), io::Error> {
+        let blocks = completed_parts
+            .into_iter()
+            .map(|part| BlockId::from(part.content_id))
+            .collect();
+
+        let block_list = BlockList { blocks };
+        let block_xml = block_list.to_xml();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(block_xml.into()),
+                true,
+                &[("comp", "blocklist")],
+            )
+            .await?;
+
+        Ok(())
+    }
+}
+
+/// Configure a connection to Microsoft Azure Blob Storage container using
+/// the specified credentials.
+///
+/// # Example
+/// ```
+/// # let ACCOUNT = "foo";
+/// # let BUCKET_NAME = "foo";
+/// # let ACCESS_KEY = "foo";
+/// # use object_store::azure::MicrosoftAzureBuilder;
+/// let azure = MicrosoftAzureBuilder::new()
+///  .with_account_name(ACCOUNT)
+///  .with_access_key_authorization(ACCESS_KEY)
+///  .with_container_name(BUCKET_NAME)
+///  .build();
+/// ```
+#[derive(Default)]
+pub struct MicrosoftAzureBuilder {
+    account_name: Option<String>,
+    access_key: Option<String>,
+    container_name: Option<String>,
+    bearer_token: Option<String>,
+    client_id: Option<String>,
+    client_secret: Option<String>,
+    tenant_id: Option<String>,
+    sas_query_pairs: Option<Vec<(String, String)>>,
+    authority_host: Option<String>,
+    use_emulator: bool,
+    retry_config: RetryConfig,
+    allow_http: bool,
+}
+
+impl Debug for MicrosoftAzureBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzureBuilder {{ account: {:?}, container_name: {:?} }}",
+            self.account_name, self.container_name
+        )
+    }
+}
+
+impl MicrosoftAzureBuilder {
+    /// Create a new [`MicrosoftAzureBuilder`] with default values.
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    /// Create an instance of [MicrosoftAzureBuilder] with values pre-populated from environment variables.
+    ///
+    /// Variables extracted from environment:
+    /// * AZURE_STORAGE_ACCOUNT_NAME: storage account name
+    /// * AZURE_STORAGE_ACCOUNT_KEY: storage account master key
+    /// * AZURE_STORAGE_ACCESS_KEY: alias for AZURE_STORAGE_ACCOUNT_KEY
+    /// * AZURE_STORAGE_CLIENT_ID -> client id for service principal authorization
+    /// * AZURE_STORAGE_CLIENT_SECRET -> client secret for service principal authorization
+    /// * AZURE_STORAGE_TENANT_ID -> tenant id used in oauth flows
+    /// # Example
+    /// ```
+    /// use object_store::azure::MicrosoftAzureBuilder;
+    ///
+    /// let azure = MicrosoftAzureBuilder::from_env()
+    ///     .with_container_name("foo")
+    ///     .build();
+    /// ```
+    pub fn from_env() -> Self {
+        let mut builder = Self::default();
+
+        if let Ok(account_name) = std::env::var("AZURE_STORAGE_ACCOUNT_NAME") {
+            builder.account_name = Some(account_name);
+        }
+
+        if let Ok(access_key) = std::env::var("AZURE_STORAGE_ACCOUNT_KEY") {
+            builder.access_key = Some(access_key);
+        } else if let Ok(access_key) = std::env::var("AZURE_STORAGE_ACCESS_KEY") {
+            builder.access_key = Some(access_key);
+        }
+
+        if let Ok(client_id) = std::env::var("AZURE_STORAGE_CLIENT_ID") {
+            builder.client_id = Some(client_id);
+        }
+
+        if let Ok(client_secret) = std::env::var("AZURE_STORAGE_CLIENT_SECRET") {
+            builder.client_secret = Some(client_secret);
+        }
+
+        if let Ok(tenant_id) = std::env::var("AZURE_STORAGE_TENANT_ID") {
+            builder.tenant_id = Some(tenant_id);
+        }
+
+        builder
+    }
+
+    /// Set the Azure Account (required)
+    pub fn with_account_name(mut self, account: impl Into<String>) -> Self {

Review Comment:
   This used to be called `with_account`?



##########
object_store/src/azure/mod.rs:
##########
@@ -0,0 +1,711 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for Azure blob storage
+//!
+//! ## Streaming uploads
+//!
+//! [ObjectStore::put_multipart] will upload data in blocks and write a blob from those
+//! blocks. Data is buffered internally to make blocks of at least 5MB and blocks
+//! are uploaded concurrently.
+//!
+//! [ObjectStore::abort_multipart] is a no-op, since Azure Blob Store doesn't provide
+//! a way to drop old blocks. Instead unused blocks are automatically cleaned up
+//! after 7 days.
+use self::client::{BlockId, BlockList};
+use crate::{
+    multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart},
+    path::Path,
+    GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result, RetryConfig,
+};
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{TimeZone, Utc};
+use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use snafu::{ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use url::Url;
+
+pub use credential::authority_hosts;
+
+mod client;
+mod credential;
+
+/// The well-known account used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT: &str = "devstoreaccount1";
+
+/// The well-known account key used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT_KEY: &str =
+    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+
+/// A specialized `Error` for Azure object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+    #[snafu(display("Last-Modified Header missing from response"))]
+    MissingLastModified,
+
+    #[snafu(display("Content-Length Header missing from response"))]
+    MissingContentLength,
+
+    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+    InvalidLastModified {
+        last_modified: String,
+        source: chrono::ParseError,
+    },
+
+    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+    InvalidContentLength {
+        content_length: String,
+        source: std::num::ParseIntError,
+    },
+
+    #[snafu(display("Received header containing non-ASCII data"))]
+    BadHeader { source: reqwest::header::ToStrError },
+
+    #[snafu(display("Unable parse source url. Url: {}, Error: {}", url, source))]
+    UnableToParseUrl {
+        source: url::ParseError,
+        url: String,
+    },
+
+    #[snafu(display(
+        "Unable parse emulator url {}={}, Error: {}",
+        env_name,
+        env_value,
+        source
+    ))]
+    UnableToParseEmulatorUrl {
+        env_name: String,
+        env_value: String,
+        source: url::ParseError,
+    },
+
+    #[snafu(display("Account must be specified"))]
+    MissingAccount {},
+
+    #[snafu(display("Container name must be specified"))]
+    MissingContainerName {},
+
+    #[snafu(display("At least one authorization option must be specified"))]
+    MissingCredentials {},
+}
+
+impl From<Error> for super::Error {
+    fn from(source: Error) -> Self {
+        Self::Generic {
+            store: "MicrosoftAzure",
+            source: Box::new(source),
+        }
+    }
+}
+
+/// Interface for [Microsoft Azure Blob Storage](https://azure.microsoft.com/en-us/services/storage/blobs/).
+#[derive(Debug)]
+pub struct MicrosoftAzure {
+    client: Arc<client::AzureClient>,
+}
+
+impl std::fmt::Display for MicrosoftAzure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzure {{ account: {}, container: {} }}",
+            self.client.config().account,
+            self.client.config().container
+        )
+    }
+}
+
+#[async_trait]
+impl ObjectStore for MicrosoftAzure {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+        self.client
+            .put_request(location, Some(bytes), false, &())
+            .await?;
+        Ok(())
+    }
+
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        let inner = AzureMultiPartUpload {
+            client: Arc::clone(&self.client),
+            location: location.to_owned(),
+        };
+        Ok((String::new(), Box::new(CloudMultiPartUpload::new(inner, 8))))
+    }
+
+    async fn abort_multipart(
+        &self,
+        _location: &Path,
+        _multipart_id: &MultipartId,
+    ) -> Result<()> {
+        // There is no way to drop blocks that have been uploaded. Instead, they simply
+        // expire in 7 days.
+        Ok(())
+    }
+
+    async fn get(&self, location: &Path) -> Result<GetResult> {
+        let response = self.client.get_request(location, None, false).await?;
+        let stream = response
+            .bytes_stream()
+            .map_err(|source| crate::Error::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(source),
+            })
+            .boxed();
+
+        Ok(GetResult::Stream(stream))
+    }
+
+    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+        let bytes = self
+            .client
+            .get_request(location, Some(range), false)
+            .await?
+            .bytes()
+            .await
+            .map_err(|source| client::Error::GetRequest {
+                source,
+                path: location.to_string(),
+            })?;
+        Ok(bytes)
+    }
+
+    async fn head(&self, location: &Path) -> Result<ObjectMeta> {
+        use reqwest::header::{CONTENT_LENGTH, LAST_MODIFIED};
+
+        // Extract meta from headers
+        // https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties
+        let response = self.client.get_request(location, None, true).await?;
+        let headers = response.headers();
+
+        let last_modified = headers
+            .get(LAST_MODIFIED)
+            .ok_or(Error::MissingLastModified)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let last_modified = Utc
+            .datetime_from_str(last_modified, credential::RFC1123_FMT)
+            .context(InvalidLastModifiedSnafu { last_modified })?;
+
+        let content_length = headers
+            .get(CONTENT_LENGTH)
+            .ok_or(Error::MissingContentLength)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let content_length = content_length
+            .parse()
+            .context(InvalidContentLengthSnafu { content_length })?;
+
+        Ok(ObjectMeta {
+            location: location.clone(),
+            last_modified,
+            size: content_length,
+        })
+    }
+
+    async fn delete(&self, location: &Path) -> Result<()> {
+        self.client.delete_request(location, &()).await
+    }
+
+    async fn list(
+        &self,
+        prefix: Option<&Path>,
+    ) -> Result<BoxStream<'_, Result<ObjectMeta>>> {
+        let stream = self
+            .client
+            .list_paginated(prefix, false)
+            .map_ok(|r| futures::stream::iter(r.objects.into_iter().map(Ok)))
+            .try_flatten()
+            .boxed();
+
+        Ok(stream)
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        let mut stream = self.client.list_paginated(prefix, true);
+
+        let mut common_prefixes = BTreeSet::new();
+        let mut objects = Vec::new();
+
+        while let Some(result) = stream.next().await {
+            let response = result?;
+            common_prefixes.extend(response.common_prefixes.into_iter());
+            objects.extend(response.objects.into_iter());
+        }
+
+        Ok(ListResult {
+            common_prefixes: common_prefixes.into_iter().collect(),
+            objects,
+        })
+    }
+
+    async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, true).await
+    }
+
+    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, false).await
+    }
+}
+
+// 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
+// complete -> PUT block list
+// abort -> No equivalent; blocks are simply dropped after 7 days
+#[derive(Debug, Clone)]
+struct AzureMultiPartUpload {
+    client: Arc<client::AzureClient>,
+    location: Path,
+}
+
+#[async_trait]
+impl CloudMultiPartUploadImpl for AzureMultiPartUpload {
+    async fn put_multipart_part(
+        &self,
+        buf: Vec<u8>,
+        part_idx: usize,
+    ) -> Result<UploadPart, io::Error> {
+        let content_id = format!("{:20}", part_idx);
+        let block_id: BlockId = content_id.clone().into();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(buf.into()),
+                true,
+                &[("comp", "block"), ("blockid", &base64::encode(block_id))],
+            )
+            .await?;
+
+        Ok(UploadPart { content_id })
+    }
+
+    async fn complete(&self, completed_parts: Vec<UploadPart>) -> Result<(), io::Error> {
+        let blocks = completed_parts
+            .into_iter()
+            .map(|part| BlockId::from(part.content_id))
+            .collect();
+
+        let block_list = BlockList { blocks };
+        let block_xml = block_list.to_xml();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(block_xml.into()),
+                true,
+                &[("comp", "blocklist")],
+            )
+            .await?;
+
+        Ok(())
+    }
+}
+
+/// Configure a connection to Microsoft Azure Blob Storage container using
+/// the specified credentials.
+///
+/// # Example
+/// ```
+/// # let ACCOUNT = "foo";
+/// # let BUCKET_NAME = "foo";
+/// # let ACCESS_KEY = "foo";
+/// # use object_store::azure::MicrosoftAzureBuilder;
+/// let azure = MicrosoftAzureBuilder::new()
+///  .with_account_name(ACCOUNT)
+///  .with_access_key_authorization(ACCESS_KEY)
+///  .with_container_name(BUCKET_NAME)
+///  .build();
+/// ```
+#[derive(Default)]
+pub struct MicrosoftAzureBuilder {
+    account_name: Option<String>,
+    access_key: Option<String>,
+    container_name: Option<String>,
+    bearer_token: Option<String>,
+    client_id: Option<String>,
+    client_secret: Option<String>,
+    tenant_id: Option<String>,
+    sas_query_pairs: Option<Vec<(String, String)>>,
+    authority_host: Option<String>,
+    use_emulator: bool,
+    retry_config: RetryConfig,
+    allow_http: bool,
+}
+
+impl Debug for MicrosoftAzureBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzureBuilder {{ account: {:?}, container_name: {:?} }}",
+            self.account_name, self.container_name
+        )
+    }
+}
+
+impl MicrosoftAzureBuilder {
+    /// Create a new [`MicrosoftAzureBuilder`] with default values.
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    /// Create an instance of [MicrosoftAzureBuilder] with values pre-populated from environment variables.
+    ///
+    /// Variables extracted from environment:
+    /// * AZURE_STORAGE_ACCOUNT_NAME: storage account name
+    /// * AZURE_STORAGE_ACCOUNT_KEY: storage account master key
+    /// * AZURE_STORAGE_ACCESS_KEY: alias for AZURE_STORAGE_ACCOUNT_KEY
+    /// * AZURE_STORAGE_CLIENT_ID -> client id for service principal authorization
+    /// * AZURE_STORAGE_CLIENT_SECRET -> client secret for service principal authorization
+    /// * AZURE_STORAGE_TENANT_ID -> tenant id used in oauth flows
+    /// # Example
+    /// ```
+    /// use object_store::azure::MicrosoftAzureBuilder;
+    ///
+    /// let azure = MicrosoftAzureBuilder::from_env()
+    ///     .with_container_name("foo")
+    ///     .build();
+    /// ```
+    pub fn from_env() -> Self {

Review Comment:
   This is a nice addition



##########
object_store/src/lib.rs:
##########
@@ -565,26 +563,16 @@ mod tests {
         let out_of_range = 200..300;
         let out_of_range_result = storage.get_range(&location, out_of_range).await;
 
-        if store_str.starts_with("MicrosoftAzureEmulator") {

Review Comment:
   :tada: 



##########
object_store/src/azure/mod.rs:
##########
@@ -0,0 +1,711 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! An object store implementation for Azure blob storage
+//!
+//! ## Streaming uploads
+//!
+//! [ObjectStore::put_multipart] will upload data in blocks and write a blob from those
+//! blocks. Data is buffered internally to make blocks of at least 5MB and blocks
+//! are uploaded concurrently.
+//!
+//! [ObjectStore::abort_multipart] is a no-op, since Azure Blob Store doesn't provide
+//! a way to drop old blocks. Instead unused blocks are automatically cleaned up
+//! after 7 days.
+use self::client::{BlockId, BlockList};
+use crate::{
+    multipart::{CloudMultiPartUpload, CloudMultiPartUploadImpl, UploadPart},
+    path::Path,
+    GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result, RetryConfig,
+};
+use async_trait::async_trait;
+use bytes::Bytes;
+use chrono::{TimeZone, Utc};
+use futures::{stream::BoxStream, StreamExt, TryStreamExt};
+use snafu::{ResultExt, Snafu};
+use std::collections::BTreeSet;
+use std::fmt::{Debug, Formatter};
+use std::io;
+use std::ops::Range;
+use std::sync::Arc;
+use tokio::io::AsyncWrite;
+use url::Url;
+
+pub use credential::authority_hosts;
+
+mod client;
+mod credential;
+
+/// The well-known account used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT: &str = "devstoreaccount1";
+
+/// The well-known account key used by Azurite and the legacy Azure Storage Emulator.
+/// <https://docs.microsoft.com/azure/storage/common/storage-use-azurite#well-known-storage-account-and-key>
+const EMULATOR_ACCOUNT_KEY: &str =
+    "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
+
+/// A specialized `Error` for Azure object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+enum Error {
+    #[snafu(display("Last-Modified Header missing from response"))]
+    MissingLastModified,
+
+    #[snafu(display("Content-Length Header missing from response"))]
+    MissingContentLength,
+
+    #[snafu(display("Invalid last modified '{}': {}", last_modified, source))]
+    InvalidLastModified {
+        last_modified: String,
+        source: chrono::ParseError,
+    },
+
+    #[snafu(display("Invalid content length '{}': {}", content_length, source))]
+    InvalidContentLength {
+        content_length: String,
+        source: std::num::ParseIntError,
+    },
+
+    #[snafu(display("Received header containing non-ASCII data"))]
+    BadHeader { source: reqwest::header::ToStrError },
+
+    #[snafu(display("Unable parse source url. Url: {}, Error: {}", url, source))]
+    UnableToParseUrl {
+        source: url::ParseError,
+        url: String,
+    },
+
+    #[snafu(display(
+        "Unable parse emulator url {}={}, Error: {}",
+        env_name,
+        env_value,
+        source
+    ))]
+    UnableToParseEmulatorUrl {
+        env_name: String,
+        env_value: String,
+        source: url::ParseError,
+    },
+
+    #[snafu(display("Account must be specified"))]
+    MissingAccount {},
+
+    #[snafu(display("Container name must be specified"))]
+    MissingContainerName {},
+
+    #[snafu(display("At least one authorization option must be specified"))]
+    MissingCredentials {},
+}
+
+impl From<Error> for super::Error {
+    fn from(source: Error) -> Self {
+        Self::Generic {
+            store: "MicrosoftAzure",
+            source: Box::new(source),
+        }
+    }
+}
+
+/// Interface for [Microsoft Azure Blob Storage](https://azure.microsoft.com/en-us/services/storage/blobs/).
+#[derive(Debug)]
+pub struct MicrosoftAzure {
+    client: Arc<client::AzureClient>,
+}
+
+impl std::fmt::Display for MicrosoftAzure {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzure {{ account: {}, container: {} }}",
+            self.client.config().account,
+            self.client.config().container
+        )
+    }
+}
+
+#[async_trait]
+impl ObjectStore for MicrosoftAzure {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+        self.client
+            .put_request(location, Some(bytes), false, &())
+            .await?;
+        Ok(())
+    }
+
+    async fn put_multipart(
+        &self,
+        location: &Path,
+    ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        let inner = AzureMultiPartUpload {
+            client: Arc::clone(&self.client),
+            location: location.to_owned(),
+        };
+        Ok((String::new(), Box::new(CloudMultiPartUpload::new(inner, 8))))
+    }
+
+    async fn abort_multipart(
+        &self,
+        _location: &Path,
+        _multipart_id: &MultipartId,
+    ) -> Result<()> {
+        // There is no way to drop blocks that have been uploaded. Instead, they simply
+        // expire in 7 days.
+        Ok(())
+    }
+
+    async fn get(&self, location: &Path) -> Result<GetResult> {
+        let response = self.client.get_request(location, None, false).await?;
+        let stream = response
+            .bytes_stream()
+            .map_err(|source| crate::Error::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(source),
+            })
+            .boxed();
+
+        Ok(GetResult::Stream(stream))
+    }
+
+    async fn get_range(&self, location: &Path, range: Range<usize>) -> Result<Bytes> {
+        let bytes = self
+            .client
+            .get_request(location, Some(range), false)
+            .await?
+            .bytes()
+            .await
+            .map_err(|source| client::Error::GetRequest {
+                source,
+                path: location.to_string(),
+            })?;
+        Ok(bytes)
+    }
+
+    async fn head(&self, location: &Path) -> Result<ObjectMeta> {
+        use reqwest::header::{CONTENT_LENGTH, LAST_MODIFIED};
+
+        // Extract meta from headers
+        // https://docs.microsoft.com/en-us/rest/api/storageservices/get-blob-properties
+        let response = self.client.get_request(location, None, true).await?;
+        let headers = response.headers();
+
+        let last_modified = headers
+            .get(LAST_MODIFIED)
+            .ok_or(Error::MissingLastModified)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let last_modified = Utc
+            .datetime_from_str(last_modified, credential::RFC1123_FMT)
+            .context(InvalidLastModifiedSnafu { last_modified })?;
+
+        let content_length = headers
+            .get(CONTENT_LENGTH)
+            .ok_or(Error::MissingContentLength)?
+            .to_str()
+            .context(BadHeaderSnafu)?;
+        let content_length = content_length
+            .parse()
+            .context(InvalidContentLengthSnafu { content_length })?;
+
+        Ok(ObjectMeta {
+            location: location.clone(),
+            last_modified,
+            size: content_length,
+        })
+    }
+
+    async fn delete(&self, location: &Path) -> Result<()> {
+        self.client.delete_request(location, &()).await
+    }
+
+    async fn list(
+        &self,
+        prefix: Option<&Path>,
+    ) -> Result<BoxStream<'_, Result<ObjectMeta>>> {
+        let stream = self
+            .client
+            .list_paginated(prefix, false)
+            .map_ok(|r| futures::stream::iter(r.objects.into_iter().map(Ok)))
+            .try_flatten()
+            .boxed();
+
+        Ok(stream)
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
+        let mut stream = self.client.list_paginated(prefix, true);
+
+        let mut common_prefixes = BTreeSet::new();
+        let mut objects = Vec::new();
+
+        while let Some(result) = stream.next().await {
+            let response = result?;
+            common_prefixes.extend(response.common_prefixes.into_iter());
+            objects.extend(response.objects.into_iter());
+        }
+
+        Ok(ListResult {
+            common_prefixes: common_prefixes.into_iter().collect(),
+            objects,
+        })
+    }
+
+    async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, true).await
+    }
+
+    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
+        self.client.copy_request(from, to, false).await
+    }
+}
+
+// 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
+// complete -> PUT block list
+// abort -> No equivalent; blocks are simply dropped after 7 days
+#[derive(Debug, Clone)]
+struct AzureMultiPartUpload {
+    client: Arc<client::AzureClient>,
+    location: Path,
+}
+
+#[async_trait]
+impl CloudMultiPartUploadImpl for AzureMultiPartUpload {
+    async fn put_multipart_part(
+        &self,
+        buf: Vec<u8>,
+        part_idx: usize,
+    ) -> Result<UploadPart, io::Error> {
+        let content_id = format!("{:20}", part_idx);
+        let block_id: BlockId = content_id.clone().into();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(buf.into()),
+                true,
+                &[("comp", "block"), ("blockid", &base64::encode(block_id))],
+            )
+            .await?;
+
+        Ok(UploadPart { content_id })
+    }
+
+    async fn complete(&self, completed_parts: Vec<UploadPart>) -> Result<(), io::Error> {
+        let blocks = completed_parts
+            .into_iter()
+            .map(|part| BlockId::from(part.content_id))
+            .collect();
+
+        let block_list = BlockList { blocks };
+        let block_xml = block_list.to_xml();
+
+        self.client
+            .put_request(
+                &self.location,
+                Some(block_xml.into()),
+                true,
+                &[("comp", "blocklist")],
+            )
+            .await?;
+
+        Ok(())
+    }
+}
+
+/// Configure a connection to Microsoft Azure Blob Storage container using
+/// the specified credentials.
+///
+/// # Example
+/// ```
+/// # let ACCOUNT = "foo";
+/// # let BUCKET_NAME = "foo";
+/// # let ACCESS_KEY = "foo";
+/// # use object_store::azure::MicrosoftAzureBuilder;
+/// let azure = MicrosoftAzureBuilder::new()
+///  .with_account_name(ACCOUNT)
+///  .with_access_key_authorization(ACCESS_KEY)
+///  .with_container_name(BUCKET_NAME)
+///  .build();
+/// ```
+#[derive(Default)]
+pub struct MicrosoftAzureBuilder {
+    account_name: Option<String>,
+    access_key: Option<String>,
+    container_name: Option<String>,
+    bearer_token: Option<String>,
+    client_id: Option<String>,
+    client_secret: Option<String>,
+    tenant_id: Option<String>,
+    sas_query_pairs: Option<Vec<(String, String)>>,
+    authority_host: Option<String>,
+    use_emulator: bool,
+    retry_config: RetryConfig,
+    allow_http: bool,
+}
+
+impl Debug for MicrosoftAzureBuilder {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "MicrosoftAzureBuilder {{ account: {:?}, container_name: {:?} }}",
+            self.account_name, self.container_name
+        )
+    }
+}
+
+impl MicrosoftAzureBuilder {

Review Comment:
   I think it would be good to avoid changing the signature of the builder, the move to replace the SDKs is a big enough change, we shouldn't add API breakage to the list if we can avoid it



##########
object_store/src/azure/client.rs:
##########
@@ -0,0 +1,740 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::credential::{AzureCredential, CredentialProvider};
+use crate::azure::credential::*;
+use crate::client::pagination::stream_paginated;
+use crate::client::retry::RetryExt;
+use crate::path::DELIMITER;
+use crate::util::{encode_path, format_http_range, format_prefix};
+use crate::{BoxStream, ListResult, ObjectMeta, Path, Result, RetryConfig, StreamExt};
+use bytes::{Buf, Bytes};
+use chrono::{DateTime, TimeZone, Utc};
+use itertools::Itertools;
+use reqwest::{
+    header::{HeaderValue, CONTENT_LENGTH, IF_NONE_MATCH},
+    Client as ReqwestClient, Method, Response, StatusCode,
+};
+use serde::{Deserialize, Deserializer, Serialize};
+use snafu::{ResultExt, Snafu};
+use std::collections::HashMap;
+use std::ops::Range;
+
+/// A specialized `Error` for object store-related errors
+#[derive(Debug, Snafu)]
+#[allow(missing_docs)]
+pub(crate) enum Error {
+    #[snafu(display("Error performing get request {}: {}", path, source))]
+    GetRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing put request {}: {}", path, source))]
+    PutRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing delete request {}: {}", path, source))]
+    DeleteRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing copy request {}: {}", path, source))]
+    CopyRequest {
+        source: reqwest::Error,
+        path: String,
+    },
+
+    #[snafu(display("Error performing list request: {}", source))]
+    ListRequest { source: reqwest::Error },
+
+    #[snafu(display("Error performing create multipart request: {}", source))]
+    CreateMultipartRequest { source: reqwest::Error },
+
+    #[snafu(display("Error performing complete multipart request: {}", source))]
+    CompleteMultipartRequest { source: reqwest::Error },
+
+    #[snafu(display("Got invalid list response: {}", source))]
+    InvalidListResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Got invalid multipart response: {}", source))]
+    InvalidMultipartResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Error authorizing request: {}", source))]
+    Authorization { source: crate::client::oauth::Error },
+}
+
+impl From<Error> for crate::Error {
+    fn from(err: Error) -> Self {
+        match err {
+            Error::GetRequest { source, path }
+            | Error::DeleteRequest { source, path }
+            | Error::CopyRequest { source, path }
+            | Error::PutRequest { source, path }
+                if matches!(source.status(), Some(StatusCode::NOT_FOUND)) =>
+            {
+                Self::NotFound {
+                    path,
+                    source: Box::new(source),
+                }
+            }
+            Error::CopyRequest { source, path }
+                if matches!(source.status(), Some(StatusCode::CONFLICT)) =>
+            {
+                Self::AlreadyExists {
+                    path,
+                    source: Box::new(source),
+                }
+            }
+            _ => Self::Generic {
+                store: "MicrosoftAzure",
+                source: Box::new(err),
+            },
+        }
+    }
+}
+
+/// Configuration for [AzureClient]
+#[derive(Debug)]
+pub struct AzureConfig {
+    pub account: String,
+    pub container: String,
+    pub credentials: CredentialProvider,
+    pub retry_config: RetryConfig,
+    pub allow_http: bool,
+    pub service: String,
+    pub is_emulator: bool,
+}
+
+impl AzureConfig {
+    fn path_url(&self, path: &Path) -> String {
+        if self.is_emulator {
+            format!(
+                "{}/{}/{}/{}",
+                self.service,
+                self.account,
+                self.container,
+                encode_path(path)
+            )
+        } else {
+            format!("{}/{}/{}", self.service, self.container, encode_path(path))
+        }
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct AzureClient {
+    config: AzureConfig,
+    client: ReqwestClient,
+}
+
+impl AzureClient {
+    /// create a new instance of [AzureClient]
+    pub fn new(config: AzureConfig) -> Self {
+        let client = reqwest::ClientBuilder::new()
+            .https_only(!config.allow_http)
+            .build()
+            .unwrap();
+
+        Self { config, client }
+    }
+
+    /// Returns the config
+    pub fn config(&self) -> &AzureConfig {
+        &self.config
+    }
+
+    async fn get_credential(&self) -> Result<AzureCredential> {
+        match &self.config.credentials {
+            CredentialProvider::AccessKey(key) => {
+                Ok(AzureCredential::AccessKey(key.to_owned()))
+            }
+            CredentialProvider::ClientSecret(cred) => {
+                let token = cred
+                    .fetch_token(&self.client, &self.config.retry_config)
+                    .await
+                    .context(AuthorizationSnafu)?;
+                Ok(AzureCredential::AuthorizationToken(
+                    // we do the conversion to a HeaderValue here, since it is fallible
+                    // and we wna to use it in an infallible function

Review Comment:
   Tbh an unwrap is perfectly fine, it is only fallible if the payload is non UTF-8 IIRC



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org