diff --git a/object_store/src/aws/client.rs b/object_store/src/aws/client.rs
index 8a45a9f3ac47..eb81e92fb932 100644
--- a/object_store/src/aws/client.rs
+++ b/object_store/src/aws/client.rs
@@ -21,6 +21,7 @@ use crate::aws::{
     AwsCredentialProvider, S3CopyIfNotExists, STORE, STRICT_PATH_ENCODE_SET,
 };
 use crate::client::get::GetClient;
+use crate::client::header::get_etag;
 use crate::client::list::ListClient;
 use crate::client::list_response::ListResponse;
 use crate::client::retry::RetryExt;
@@ -122,6 +123,11 @@ pub(crate) enum Error {
 
     #[snafu(display("Got invalid multipart response: {}", source))]
     InvalidMultipartResponse { source: quick_xml::de::DeError },
+
+    #[snafu(display("Unable to extract metadata from headers: {}", source))]
+    Metadata {
+        source: crate::client::header::Error,
+    },
 }
 
 impl From<Error> for crate::Error {
@@ -243,12 +249,14 @@ impl S3Client {
     }
 
     /// Make an S3 PUT request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html>
+    ///
+    /// Returns the ETag
     pub async fn put_request<T: Serialize + ?Sized + Sync>(
         &self,
         path: &Path,
         bytes: Bytes,
         query: &T,
-    ) -> Result<Response> {
+    ) -> Result<String> {
         let credential = self.get_credential().await?;
         let url = self.config.path_url(path);
         let mut builder = self.client.request(Method::PUT, url);
@@ -287,7 +295,7 @@ impl S3Client {
                 path: path.as_ref(),
             })?;
 
-        Ok(response)
+        Ok(get_etag(response.headers()).context(MetadataSnafu)?)
     }
 
     /// Make an S3 Delete request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObject.html>
diff --git a/object_store/src/aws/mod.rs b/object_store/src/aws/mod.rs
index 3ddce08002c4..785ce20faa6d 100644
--- a/object_store/src/aws/mod.rs
+++ b/object_store/src/aws/mod.rs
@@ -59,7 +59,7 @@ use crate::multipart::{PartId, PutPart, WriteMultiPart};
 use crate::signer::Signer;
 use crate::{
     ClientOptions, GetOptions, GetResult, ListResult, MultipartId, ObjectMeta,
-    ObjectStore, Path, Result, RetryConfig,
+    ObjectStore, Path, PutResult, Result, RetryConfig,
 };
 
 mod checksum;
@@ -109,12 +109,6 @@ enum Error {
     #[snafu(display("Missing SecretAccessKey"))]
     MissingSecretAccessKey,
 
-    #[snafu(display("ETag Header missing from response"))]
-    MissingEtag,
-
-    #[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,
@@ -273,9 +267,9 @@ impl Signer for AmazonS3 {
 
 #[async_trait]
 impl ObjectStore for AmazonS3 {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
-        self.client.put_request(location, bytes, &()).await?;
-        Ok(())
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
+        let e_tag = self.client.put_request(location, bytes, &()).await?;
+        Ok(PutResult { e_tag: Some(e_tag) })
     }
 
     async fn put_multipart(
@@ -368,10 +362,9 @@ struct S3MultiPartUpload {
 #[async_trait]
 impl PutPart for S3MultiPartUpload {
     async fn put_part(&self, buf: Vec<u8>, part_idx: usize) -> Result<PartId> {
-        use reqwest::header::ETAG;
         let part = (part_idx + 1).to_string();
 
-        let response = self
+        let content_id = self
             .client
             .put_request(
                 &self.location,
@@ -380,13 +373,7 @@ impl PutPart for S3MultiPartUpload {
             )
             .await?;
 
-        let etag = response.headers().get(ETAG).context(MissingEtagSnafu)?;
-
-        let etag = etag.to_str().context(BadHeaderSnafu)?;
-
-        Ok(PartId {
-            content_id: etag.to_string(),
-        })
+        Ok(PartId { content_id })
     }
 
     async fn complete(&self, completed_parts: Vec<PartId>) -> Result<()> {
diff --git a/object_store/src/azure/mod.rs b/object_store/src/azure/mod.rs
index 190b73bf9490..039a877664bb 100644
--- a/object_store/src/azure/mod.rs
+++ b/object_store/src/azure/mod.rs
@@ -31,7 +31,7 @@ use crate::{
     multipart::{PartId, PutPart, WriteMultiPart},
     path::Path,
     ClientOptions, GetOptions, GetResult, ListResult, MultipartId, ObjectMeta,
-    ObjectStore, Result, RetryConfig,
+    ObjectStore, PutResult, Result, RetryConfig,
 };
 use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
@@ -62,6 +62,7 @@ mod credential;
 /// [`CredentialProvider`] for [`MicrosoftAzure`]
 pub type AzureCredentialProvider =
     Arc<dyn CredentialProvider<Credential = AzureCredential>>;
+use crate::client::header::get_etag;
 pub use credential::AzureCredential;
 
 const STORE: &str = "MicrosoftAzure";
@@ -81,9 +82,6 @@ const MSI_ENDPOINT_ENV_KEY: &str = "IDENTITY_ENDPOINT";
 #[derive(Debug, Snafu)]
 #[allow(missing_docs)]
 enum Error {
-    #[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,
@@ -126,8 +124,10 @@ enum Error {
     #[snafu(display("Configuration key: '{}' is not known.", key))]
     UnknownConfigurationKey { key: String },
 
-    #[snafu(display("ETag Header missing from response"))]
-    MissingEtag,
+    #[snafu(display("Unable to extract metadata from headers: {}", source))]
+    Metadata {
+        source: crate::client::header::Error,
+    },
 }
 
 impl From<Error> for super::Error {
@@ -170,11 +170,13 @@ impl std::fmt::Display for MicrosoftAzure {
 
 #[async_trait]
 impl ObjectStore for MicrosoftAzure {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
-        self.client
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
+        let response = self
+            .client
             .put_request(location, Some(bytes), false, &())
             .await?;
-        Ok(())
+        let e_tag = Some(get_etag(response.headers()).context(MetadataSnafu)?);
+        Ok(PutResult { e_tag })
     }
 
     async fn put_multipart(
diff --git a/object_store/src/chunked.rs b/object_store/src/chunked.rs
index 008dec679413..aedb99831703 100644
--- a/object_store/src/chunked.rs
+++ b/object_store/src/chunked.rs
@@ -30,6 +30,7 @@ use tokio::io::AsyncWrite;
 use crate::path::Path;
 use crate::{
     GetOptions, GetResult, GetResultPayload, ListResult, ObjectMeta, ObjectStore,
+    PutResult,
 };
 use crate::{MultipartId, Result};
 
@@ -62,7 +63,7 @@ impl Display for ChunkedStore {
 
 #[async_trait]
 impl ObjectStore for ChunkedStore {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
         self.inner.put(location, bytes).await
     }
 
diff --git a/object_store/src/client/header.rs b/object_store/src/client/header.rs
index 6499eff5aebe..17f83a2ba8c8 100644
--- a/object_store/src/client/header.rs
+++ b/object_store/src/client/header.rs
@@ -64,6 +64,12 @@ pub enum Error {
     },
 }
 
+/// Extracts an etag from the provided [`HeaderMap`]
+pub fn get_etag(headers: &HeaderMap) -> Result<String, Error> {
+    let e_tag = headers.get(ETAG).ok_or(Error::MissingEtag)?;
+    Ok(e_tag.to_str().context(BadHeaderSnafu)?.to_string())
+}
+
 /// Extracts [`ObjectMeta`] from the provided [`HeaderMap`]
 pub fn header_meta(
     location: &Path,
@@ -81,13 +87,10 @@ pub fn header_meta(
         None => Utc.timestamp_nanos(0),
     };
 
-    let e_tag = match headers.get(ETAG) {
-        Some(e_tag) => {
-            let e_tag = e_tag.to_str().context(BadHeaderSnafu)?;
-            Some(e_tag.to_string())
-        }
-        None if cfg.etag_required => return Err(Error::MissingEtag),
-        None => None,
+    let e_tag = match get_etag(headers) {
+        Ok(e_tag) => Some(e_tag),
+        Err(Error::MissingEtag) if !cfg.etag_required => None,
+        Err(e) => return Err(e),
     };
 
     let content_length = headers
diff --git a/object_store/src/gcp/mod.rs b/object_store/src/gcp/mod.rs
index a75527fe7b9f..4e2e108df657 100644
--- a/object_store/src/gcp/mod.rs
+++ b/object_store/src/gcp/mod.rs
@@ -54,7 +54,7 @@ use crate::{
     multipart::{PartId, PutPart, WriteMultiPart},
     path::{Path, DELIMITER},
     ClientOptions, GetOptions, GetResult, ListResult, MultipartId, ObjectMeta,
-    ObjectStore, Result, RetryConfig,
+    ObjectStore, PutResult, Result, RetryConfig,
 };
 
 use credential::{InstanceCredentialProvider, ServiceAccountCredentials};
@@ -65,6 +65,7 @@ const STORE: &str = "GCS";
 
 /// [`CredentialProvider`] for [`GoogleCloudStorage`]
 pub type GcpCredentialProvider = Arc<dyn CredentialProvider<Credential = GcpCredential>>;
+use crate::client::header::get_etag;
 use crate::gcp::credential::{ApplicationDefaultCredentials, DEFAULT_GCS_BASE_URL};
 pub use credential::GcpCredential;
 
@@ -155,11 +156,10 @@ enum Error {
     #[snafu(display("Configuration key: '{}' is not known.", key))]
     UnknownConfigurationKey { key: String },
 
-    #[snafu(display("ETag Header missing from response"))]
-    MissingEtag,
-
-    #[snafu(display("Received header containing non-ASCII data"))]
-    BadHeader { source: header::ToStrError },
+    #[snafu(display("Unable to extract metadata from headers: {}", source))]
+    Metadata {
+        source: crate::client::header::Error,
+    },
 }
 
 impl From<Error> for super::Error {
@@ -247,7 +247,14 @@ impl GoogleCloudStorageClient {
     }
 
     /// Perform a put request <https://cloud.google.com/storage/docs/xml-api/put-object-upload>
-    async fn put_request(&self, path: &Path, payload: Bytes) -> Result<()> {
+    ///
+    /// Returns the new ETag
+    async fn put_request<T: Serialize + ?Sized + Sync>(
+        &self,
+        path: &Path,
+        payload: Bytes,
+        query: &T,
+    ) -> Result<String> {
         let credential = self.get_credential().await?;
         let url = self.object_url(path);
 
@@ -256,8 +263,10 @@ impl GoogleCloudStorageClient {
             .get_content_type(path)
             .unwrap_or("application/octet-stream");
 
-        self.client
+        let response = self
+            .client
             .request(Method::PUT, url)
+            .query(query)
             .bearer_auth(&credential.bearer)
             .header(header::CONTENT_TYPE, content_type)
             .header(header::CONTENT_LENGTH, payload.len())
@@ -268,7 +277,7 @@ impl GoogleCloudStorageClient {
                 path: path.as_ref(),
             })?;
 
-        Ok(())
+        Ok(get_etag(response.headers()).context(MetadataSnafu)?)
     }
 
     /// Initiate a multi-part upload <https://cloud.google.com/storage/docs/xml-api/post-object-multipart>
@@ -469,7 +478,7 @@ impl ListClient for GoogleCloudStorageClient {
 
 struct GCSMultipartUpload {
     client: Arc<GoogleCloudStorageClient>,
-    encoded_path: String,
+    path: Path,
     multipart_id: MultipartId,
 }
 
@@ -478,38 +487,17 @@ impl PutPart for GCSMultipartUpload {
     /// Upload an object part <https://cloud.google.com/storage/docs/xml-api/put-object-multipart>
     async fn put_part(&self, buf: Vec<u8>, part_idx: usize) -> Result<PartId> {
         let upload_id = self.multipart_id.clone();
-        let url = format!(
-            "{}/{}/{}",
-            self.client.base_url, self.client.bucket_name_encoded, self.encoded_path
-        );
-
-        let credential = self.client.get_credential().await?;
-
-        let response = self
+        let content_id = self
             .client
-            .client
-            .request(Method::PUT, &url)
-            .bearer_auth(&credential.bearer)
-            .query(&[
-                ("partNumber", format!("{}", part_idx + 1)),
-                ("uploadId", upload_id),
-            ])
-            .header(header::CONTENT_TYPE, "application/octet-stream")
-            .header(header::CONTENT_LENGTH, format!("{}", buf.len()))
-            .body(buf)
-            .send_retry(&self.client.retry_config)
-            .await
-            .context(PutRequestSnafu {
-                path: &self.encoded_path,
-            })?;
-
-        let content_id = response
-            .headers()
-            .get("ETag")
-            .context(MissingEtagSnafu)?
-            .to_str()
-            .context(BadHeaderSnafu)?
-            .to_string();
+            .put_request(
+                &self.path,
+                buf.into(),
+                &[
+                    ("partNumber", format!("{}", part_idx + 1)),
+                    ("uploadId", upload_id),
+                ],
+            )
+            .await?;
 
         Ok(PartId { content_id })
     }
@@ -517,10 +505,7 @@ impl PutPart for GCSMultipartUpload {
     /// Complete a multipart upload <https://cloud.google.com/storage/docs/xml-api/post-object-complete>
     async fn complete(&self, completed_parts: Vec<PartId>) -> Result<()> {
         let upload_id = self.multipart_id.clone();
-        let url = format!(
-            "{}/{}/{}",
-            self.client.base_url, self.client.bucket_name_encoded, self.encoded_path
-        );
+        let url = self.client.object_url(&self.path);
 
         let parts = completed_parts
             .into_iter()
@@ -550,7 +535,7 @@ impl PutPart for GCSMultipartUpload {
             .send_retry(&self.client.retry_config)
             .await
             .context(PostRequestSnafu {
-                path: &self.encoded_path,
+                path: self.path.as_ref(),
             })?;
 
         Ok(())
@@ -559,8 +544,9 @@ impl PutPart for GCSMultipartUpload {
 
 #[async_trait]
 impl ObjectStore for GoogleCloudStorage {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
-        self.client.put_request(location, bytes).await
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
+        let e_tag = self.client.put_request(location, bytes, &()).await?;
+        Ok(PutResult { e_tag: Some(e_tag) })
     }
 
     async fn put_multipart(
@@ -569,12 +555,9 @@ impl ObjectStore for GoogleCloudStorage {
     ) -> Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
         let upload_id = self.client.multipart_initiate(location).await?;
 
-        let encoded_path =
-            percent_encode(location.to_string().as_bytes(), NON_ALPHANUMERIC).to_string();
-
         let inner = GCSMultipartUpload {
             client: Arc::clone(&self.client),
-            encoded_path,
+            path: location.clone(),
             multipart_id: upload_id.clone(),
         };
 
diff --git a/object_store/src/http/client.rs b/object_store/src/http/client.rs
index b2a6ac0aa34a..4c2a7fcf8db3 100644
--- a/object_store/src/http/client.rs
+++ b/object_store/src/http/client.rs
@@ -160,7 +160,7 @@ impl Client {
         Ok(())
     }
 
-    pub async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+    pub async fn put(&self, location: &Path, bytes: Bytes) -> Result<Response> {
         let mut retry = false;
         loop {
             let url = self.path_url(location);
@@ -170,7 +170,7 @@ impl Client {
             }
 
             match builder.send_retry(&self.retry_config).await {
-                Ok(_) => return Ok(()),
+                Ok(response) => return Ok(response),
                 Err(source) => match source.status() {
                     // Some implementations return 404 instead of 409
                     Some(StatusCode::CONFLICT | StatusCode::NOT_FOUND) if !retry => {
diff --git a/object_store/src/http/mod.rs b/object_store/src/http/mod.rs
index 6ffb62358941..e2fc8801eacb 100644
--- a/object_store/src/http/mod.rs
+++ b/object_store/src/http/mod.rs
@@ -41,11 +41,12 @@ use tokio::io::AsyncWrite;
 use url::Url;
 
 use crate::client::get::GetClientExt;
+use crate::client::header::get_etag;
 use crate::http::client::Client;
 use crate::path::Path;
 use crate::{
     ClientConfigKey, ClientOptions, GetOptions, GetResult, ListResult, MultipartId,
-    ObjectMeta, ObjectStore, Result, RetryConfig,
+    ObjectMeta, ObjectStore, PutResult, Result, RetryConfig,
 };
 
 mod client;
@@ -95,8 +96,14 @@ impl std::fmt::Display for HttpStore {
 
 #[async_trait]
 impl ObjectStore for HttpStore {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
-        self.client.put(location, bytes).await
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
+        let response = self.client.put(location, bytes).await?;
+        let e_tag = match get_etag(response.headers()) {
+            Ok(e_tag) => Some(e_tag),
+            Err(crate::client::header::Error::MissingEtag) => None,
+            Err(source) => return Err(Error::Metadata { source }.into()),
+        };
+        Ok(PutResult { e_tag })
     }
 
     async fn put_multipart(
diff --git a/object_store/src/lib.rs b/object_store/src/lib.rs
index b79042e3cda8..0afa159962ff 100644
--- a/object_store/src/lib.rs
+++ b/object_store/src/lib.rs
@@ -320,7 +320,7 @@ pub trait ObjectStore: std::fmt::Display + Send + Sync + Debug + 'static {
     /// The operation is guaranteed to be atomic, it will either successfully
     /// write the entirety of `bytes` to `location`, or fail. No clients
     /// should be able to observe a partially written object
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()>;
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult>;
 
     /// Get a multi-part upload that allows writing data in chunks
     ///
@@ -555,7 +555,7 @@ macro_rules! as_ref_impl {
     ($type:ty) => {
         #[async_trait]
         impl ObjectStore for $type {
-            async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+            async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
                 self.as_ref().put(location, bytes).await
             }
 
@@ -880,6 +880,13 @@ impl GetResult {
     }
 }
 
+/// Result for a put request
+#[derive(Debug)]
+pub struct PutResult {
+    /// The unique identifier for the object
+    pub e_tag: Option<String>,
+}
+
 /// A specialized `Result` for object store-related errors
 pub type Result<T, E = Error> = std::result::Result<T, E>;
 
@@ -1717,7 +1724,7 @@ mod tests {
     async fn list_store<'a, 'b>(
         store: &'a dyn ObjectStore,
         path_str: &'b str,
-    ) -> super::Result<BoxStream<'a, super::Result<ObjectMeta>>> {
+    ) -> Result<BoxStream<'a, Result<ObjectMeta>>> {
         let path = Path::from(path_str);
         store.list(Some(&path)).await
     }
diff --git a/object_store/src/limit.rs b/object_store/src/limit.rs
index a9b8c4b05020..385599e19134 100644
--- a/object_store/src/limit.rs
+++ b/object_store/src/limit.rs
@@ -19,7 +19,7 @@
 
 use crate::{
     BoxStream, GetOptions, GetResult, GetResultPayload, ListResult, MultipartId,
-    ObjectMeta, ObjectStore, Path, Result, StreamExt,
+    ObjectMeta, ObjectStore, Path, PutResult, Result, StreamExt,
 };
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -72,7 +72,7 @@ impl<T: ObjectStore> std::fmt::Display for LimitStore<T> {
 
 #[async_trait]
 impl<T: ObjectStore> ObjectStore for LimitStore<T> {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
         let _permit = self.semaphore.acquire().await.unwrap();
         self.inner.put(location, bytes).await
     }
diff --git a/object_store/src/local.rs b/object_store/src/local.rs
index 3d4a02a1e9e9..2020c07e8add 100644
--- a/object_store/src/local.rs
+++ b/object_store/src/local.rs
@@ -20,7 +20,7 @@ use crate::{
     maybe_spawn_blocking,
     path::{absolute_path_to_url, Path},
     GetOptions, GetResult, GetResultPayload, ListResult, MultipartId, ObjectMeta,
-    ObjectStore, Result,
+    ObjectStore, PutResult, Result,
 };
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -36,6 +36,7 @@ use std::ops::Range;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Poll;
+use std::time::SystemTime;
 use std::{collections::BTreeSet, convert::TryFrom, io};
 use std::{collections::VecDeque, path::PathBuf};
 use tokio::io::AsyncWrite;
@@ -270,7 +271,7 @@ impl Config {
 
 #[async_trait]
 impl ObjectStore for LocalFileSystem {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
         let path = self.config.path_to_filesystem(location)?;
         maybe_spawn_blocking(move || {
             let (mut file, suffix) = new_staged_upload(&path)?;
@@ -282,8 +283,17 @@ impl ObjectStore for LocalFileSystem {
                 })
                 .map_err(|e| {
                     let _ = std::fs::remove_file(&staging_path); // Attempt to cleanup
-                    e.into()
-                })
+                    e
+                })?;
+
+            let metadata = file.metadata().map_err(|e| Error::Metadata {
+                source: e.into(),
+                path: path.to_string_lossy().to_string(),
+            })?;
+
+            Ok(PutResult {
+                e_tag: Some(get_etag(&metadata)),
+            })
         })
         .await
     }
@@ -961,24 +971,33 @@ fn last_modified(metadata: &Metadata) -> DateTime<Utc> {
         .into()
 }
 
-fn convert_metadata(metadata: Metadata, location: Path) -> Result<ObjectMeta> {
-    let last_modified = last_modified(&metadata);
-    let size = usize::try_from(metadata.len()).context(FileSizeOverflowedUsizeSnafu {
-        path: location.as_ref(),
-    })?;
+fn get_etag(metadata: &Metadata) -> String {
     let inode = get_inode(&metadata);
-    let mtime = last_modified.timestamp_micros();
+    let size = metadata.len();
+    let mtime = metadata
+        .modified()
+        .ok()
+        .and_then(|mtime| mtime.duration_since(SystemTime::UNIX_EPOCH).ok())
+        .unwrap_or_default()
+        .as_micros();
 
     // Use an ETag scheme based on that used by many popular HTTP servers
     // <https://httpd.apache.org/docs/2.2/mod/core.html#fileetag>
     // <https://stackoverflow.com/questions/47512043/how-etags-are-generated-and-configured>
-    let etag = format!("{inode:x}-{mtime:x}-{size:x}");
+    format!("{inode:x}-{mtime:x}-{size:x}")
+}
+
+fn convert_metadata(metadata: Metadata, location: Path) -> Result<ObjectMeta> {
+    let last_modified = last_modified(&metadata);
+    let size = usize::try_from(metadata.len()).context(FileSizeOverflowedUsizeSnafu {
+        path: location.as_ref(),
+    })?;
 
     Ok(ObjectMeta {
         location,
         last_modified,
         size,
-        e_tag: Some(etag),
+        e_tag: Some(get_etag(&metadata)),
     })
 }
 
diff --git a/object_store/src/memory.rs b/object_store/src/memory.rs
index f638ed6d7a55..26ce6cc7c591 100644
--- a/object_store/src/memory.rs
+++ b/object_store/src/memory.rs
@@ -17,7 +17,8 @@
 
 //! An in-memory object store implementation
 use crate::{
-    path::Path, GetResult, GetResultPayload, ListResult, ObjectMeta, ObjectStore, Result,
+    path::Path, GetResult, GetResultPayload, ListResult, ObjectMeta, ObjectStore,
+    PutResult, Result,
 };
 use crate::{GetOptions, MultipartId};
 use async_trait::async_trait;
@@ -106,11 +107,12 @@ struct Storage {
 type SharedStorage = Arc<RwLock<Storage>>;
 
 impl Storage {
-    fn insert(&mut self, location: &Path, bytes: Bytes) {
+    fn insert(&mut self, location: &Path, bytes: Bytes) -> usize {
         let etag = self.next_etag;
         self.next_etag += 1;
         let entry = Entry::new(bytes, Utc::now(), etag);
         self.map.insert(location.clone(), entry);
+        etag
     }
 }
 
@@ -122,9 +124,11 @@ impl std::fmt::Display for InMemory {
 
 #[async_trait]
 impl ObjectStore for InMemory {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
-        self.storage.write().insert(location, bytes);
-        Ok(())
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
+        let etag = self.storage.write().insert(location, bytes);
+        Ok(PutResult {
+            e_tag: Some(etag.to_string()),
+        })
     }
 
     async fn put_multipart(
diff --git a/object_store/src/prefix.rs b/object_store/src/prefix.rs
index 39585f73b692..10ed9188593c 100644
--- a/object_store/src/prefix.rs
+++ b/object_store/src/prefix.rs
@@ -23,7 +23,8 @@ use tokio::io::AsyncWrite;
 
 use crate::path::Path;
 use crate::{
-    GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result,
+    GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, PutResult,
+    Result,
 };
 
 #[doc(hidden)]
@@ -79,7 +80,7 @@ impl<T: ObjectStore> PrefixStore<T> {
 
 #[async_trait::async_trait]
 impl<T: ObjectStore> ObjectStore for PrefixStore<T> {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
         let full_path = self.full_path(location);
         self.inner.put(&full_path, bytes).await
     }
diff --git a/object_store/src/throttle.rs b/object_store/src/throttle.rs
index 58c476ab4530..fd2be118f31a 100644
--- a/object_store/src/throttle.rs
+++ b/object_store/src/throttle.rs
@@ -21,7 +21,8 @@ use std::ops::Range;
 use std::{convert::TryInto, sync::Arc};
 
 use crate::{
-    path::Path, GetResult, GetResultPayload, ListResult, ObjectMeta, ObjectStore, Result,
+    path::Path, GetResult, GetResultPayload, ListResult, ObjectMeta, ObjectStore,
+    PutResult, Result,
 };
 use crate::{GetOptions, MultipartId};
 use async_trait::async_trait;
@@ -147,7 +148,7 @@ impl<T: ObjectStore> std::fmt::Display for ThrottledStore<T> {
 
 #[async_trait]
 impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
-    async fn put(&self, location: &Path, bytes: Bytes) -> Result<()> {
+    async fn put(&self, location: &Path, bytes: Bytes) -> Result<PutResult> {
         sleep(self.config().wait_put_per_call).await;
 
         self.inner.put(location, bytes).await
diff --git a/object_store/tests/get_range_file.rs b/object_store/tests/get_range_file.rs
index f926e3b07f2a..3f30eea1357e 100644
--- a/object_store/tests/get_range_file.rs
+++ b/object_store/tests/get_range_file.rs
@@ -23,7 +23,7 @@ use futures::stream::BoxStream;
 use object_store::local::LocalFileSystem;
 use object_store::path::Path;
 use object_store::{
-    GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore,
+    GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, PutResult,
 };
 use std::fmt::Formatter;
 use tempfile::tempdir;
@@ -40,7 +40,7 @@ impl std::fmt::Display for MyStore {
 
 #[async_trait]
 impl ObjectStore for MyStore {
-    async fn put(&self, path: &Path, data: Bytes) -> object_store::Result<()> {
+    async fn put(&self, path: &Path, data: Bytes) -> object_store::Result<PutResult> {
         self.0.put(path, data).await
     }