feat: List metadata reuse (#577)

* feature: list-metadata-reuse

1. add more fields to DirEntry
2. record fields in HDFS
3. record fields in GCS
4. record fields in S3
5. record fields in Microsoft Cloud

Signed-off-by: ClSlaid <[email protected]>

* feature: implement extended DirEntry for fs

Signed-off-by: ClSlaid <[email protected]>

* refactor: make cargo fmt happy

Signed-off-by: ClSlaid <[email protected]>

Signed-off-by: ClSlaid <[email protected]>

Author: ClSlaid

src/lib.rs

@@ -168,7 +168,7 @@ mod tests {
         assert_eq!(80, size_of::<AccessorMetadata>());
         assert_eq!(16, size_of::<Operator>());
         assert_eq!(16, size_of::<BatchOperator>());
-        assert_eq!(48, size_of::<DirEntry>());
+        assert_eq!(128, size_of::<DirEntry>());
         assert_eq!(40, size_of::<Object>());
         assert_eq!(80, size_of::<ObjectMetadata>());
         assert_eq!(1, size_of::<ObjectMode>());

src/object.rs

@@ -908,6 +908,12 @@ pub struct DirEntry {
 
     mode: ObjectMode,
     path: String,
+
+    // metadata fields
+    etag: Option<String>,
+    content_length: Option<u64>,
+    content_md5: Option<String>,
+    last_modified: Option<OffsetDateTime>,
 }
 
 impl DirEntry {
@@ -923,6 +929,11 @@ impl DirEntry {
             acc,
             mode,
             path: path.to_string(),
+            // all set to None
+            etag: None,
+            content_length: None,
+            content_md5: None,
+            last_modified: None,
         }
     }
 
@@ -961,6 +972,46 @@ impl DirEntry {
         get_basename(&self.path)
     }
 
+    /// The ETag string of `DirEntry`'s corresponding object
+    ///
+    /// `etag` is a prefetched metadata field in `DirEntry`.
+    ///
+    /// It doesn't mean this metadata field of object doesn't exist if `etag` is `None`.
+    /// Then you have to call `DirEntry::metadata()` to get the metadata you want.
+    pub fn etag(&self) -> Option<&str> {
+        self.etag.as_deref()
+    }
+
+    /// The size of `DirEntry`'s corresponding object
+    ///
+    /// `content_length` is a prefetched metadata field in `DirEntry`.
+    ///
+    /// It doesn't mean this metadata field of object doesn't exist if `content_length` is `None`.
+    /// Then you have to call `DirEntry::metadata()` to get the metadata you want.
+    pub fn content_length(&self) -> Option<u64> {
+        self.content_length
+    }
+
+    /// The MD5 message digest of `DirEntry`'s corresponding object
+    ///
+    /// `content_md5` is a prefetched metadata field in `DirEntry`
+    ///
+    /// It doesn't mean this metadata field of object doesn't exist if `content_md5` is `None`.
+    /// Then you have to call `DirEntry::metadata()` to get the metadata you want.
+    pub fn content_md5(&self) -> Option<&str> {
+        self.content_md5.as_deref()
+    }
+
+    /// The last modified UTC datetime of `DirEntry`'s corresponding object
+    ///
+    /// `last_modified` is a prefetched metadata field in `DirEntry`
+    ///
+    /// It doesn't mean this metadata field of object doesn't exist if `last_modified` is `None`.
+    /// Then you have to call `DirEntry::metadata()` to get the metadata you want.
+    pub fn last_modified(&self) -> Option<OffsetDateTime> {
+        self.last_modified
+    }
+
     /// Fetch metadata about this dir entry.
     ///
     /// The same with [`Object::metadata()`]
@@ -971,6 +1022,26 @@ impl DirEntry {
     }
 }
 
+// implement setters for DirEntry's metadata fields
+impl DirEntry {
+    /// record the ETag of `DirEntry`'s corresponding object
+    pub(crate) fn set_etag(&mut self, etag: &str) {
+        self.etag = Some(etag.to_string())
+    }
+    /// record the last modified time of `DirEntry`'s corresponding object
+    pub(crate) fn set_last_modified(&mut self, last_modified: OffsetDateTime) {
+        self.last_modified = Some(last_modified)
+    }
+    /// record the content length of `DirEntry`'s corresponding object
+    pub(crate) fn set_content_length(&mut self, content_length: u64) {
+        self.content_length = Some(content_length)
+    }
+    /// record the content's md5 of `DirEntry`'s corresponding object
+    pub(crate) fn set_content_md5(&mut self, content_md5: &str) {
+        self.content_md5 = Some(content_md5.to_string())
+    }
+}
+
 /// DirEntry can convert into object without overhead.
 impl From<DirEntry> for Object {
     fn from(d: DirEntry) -> Self {

src/services/azblob/dir_stream.rs

@@ -11,6 +11,7 @@
 // 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 anyhow::anyhow;
 use std::future::Future;
 use std::io::Result;
 use std::pin::Pin;
@@ -25,6 +26,8 @@ use isahc::AsyncReadResponseExt;
 use log::debug;
 use quick_xml::de;
 use serde::Deserialize;
+use time::format_description::well_known::Rfc2822;
+use time::OffsetDateTime;
 
 use super::error::parse_error;
 use super::Backend;
@@ -119,11 +122,15 @@ impl futures::Stream for DirStream {
                         );
 
                         debug!(
-                            "dir object {} got entry, mode: {}, path: {}",
-                            &self.path,
-                            de.mode(),
-                            de.path()
-                        );
+                        "dir object {} got entry, mode: {}, path: {}, content length: {:?}, last modified: {:?}, content_md5: {:?}, etag: {:?}",
+                        &self.path,
+                        de.mode(),
+                        de.path(),
+                        de.content_length(),
+                        de.last_modified(),
+                        de.content_md5(),
+                        de.etag()
+                    );
                         return Poll::Ready(Some(Ok(de)));
                     }
                 };
@@ -140,17 +147,36 @@ impl futures::Stream for DirStream {
                         continue;
                     }
 
-                    let de = DirEntry::new(
+                    let mut de = DirEntry::new(
                         backend.clone(),
                         ObjectMode::FILE,
                         &backend.get_rel_path(&object.name),
                     );
 
+                    de.set_etag(object.properties.etag.as_str());
+                    de.set_content_length(object.properties.content_length);
+                    de.set_content_md5(object.properties.content_md5.as_str());
+
+                    let dt =
+                        OffsetDateTime::parse(object.properties.last_modified.as_str(), &Rfc2822)
+                            .map_err(|e| {
+                            other(ObjectError::new(
+                                "list",
+                                &self.path,
+                                anyhow!("parse last modified RFC2822 datetime: {e:?}"),
+                            ))
+                        })?;
+                    de.set_last_modified(dt);
+
                     debug!(
-                        "dir object {} got entry, mode: {}, path: {}",
+                        "dir object {} got entry, mode: {}, path: {}, content length: {:?}, last modified: {:?}, content_md5: {:?}, etag: {:?}",
                         &self.path,
                         de.mode(),
                         de.path(),
+                        de.content_length(),
+                        de.last_modified(),
+                        de.content_md5(),
+                        de.etag()
                     );
                     return Poll::Ready(Some(Ok(de)));
                 }
@@ -200,6 +226,11 @@ struct Blob {
 struct Properties {
     #[serde(rename = "Content-Length")]
     content_length: u64,
+    #[serde(rename = "Last-Modified")]
+    last_modified: String,
+    #[serde(rename = "Content-MD5")]
+    content_md5: String,
+    etag: String,
 }
 
 #[cfg(test)]
@@ -317,6 +348,42 @@ mod tests {
                 .collect::<Vec<u64>>(),
             [3485277, 2471869, 1259677]
         );
+        assert_eq!(
+            out.blobs
+                .blob
+                .iter()
+                .map(|v| v.properties.content_md5.clone())
+                .collect::<Vec<String>>(),
+            [
+                "llJ/+jOlx5GdA1sL7SdKuw==".to_string(),
+                "xmgUltSnopLSJOukgCHFtg==".to_string(),
+                "AxTiFXHwrXKaZC5b7ZRybw==".to_string()
+            ]
+        );
+        assert_eq!(
+            out.blobs
+                .blob
+                .iter()
+                .map(|v| v.properties.last_modified.clone())
+                .collect::<Vec<String>>(),
+            [
+                "Sun, 20 Mar 2022 11:29:03 GMT".to_string(),
+                "Tue, 29 Mar 2022 01:54:07 GMT".to_string(),
+                "Sun, 20 Mar 2022 11:31:57 GMT".to_string()
+            ]
+        );
+        assert_eq!(
+            out.blobs
+                .blob
+                .iter()
+                .map(|v| v.properties.etag.clone())
+                .collect::<Vec<String>>(),
+            [
+                "0x8DA0A64D66790C3".to_string(),
+                "0x8DA112702D88FE4".to_string(),
+                "0x8DA0A653DC82981".to_string()
+            ]
+        );
         assert_eq!(
             out.blobs
                 .blob_prefix

src/services/fs/dir_stream.rs

@@ -20,6 +20,7 @@ use std::task::Poll;
 
 use log::debug;
 use log::error;
+use time::OffsetDateTime;
 
 use super::error::parse_io_error;
 use super::Backend;
@@ -65,7 +66,7 @@ impl futures::Stream for DirStream {
                 // the target file type.
                 let file_type = de.file_type()?;
 
-                let d = if file_type.is_file() {
+                let mut d = if file_type.is_file() {
                     DirEntry::new(self.backend.clone(), ObjectMode::FILE, &path)
                 } else if file_type.is_dir() {
                     // Make sure we are returning the correct path.
@@ -78,12 +79,26 @@ impl futures::Stream for DirStream {
                     DirEntry::new(self.backend.clone(), ObjectMode::Unknown, &path)
                 };
 
+                // metadata may not available on all platforms, it's ok not setting it here
+                if let Ok(metadata) = de.metadata() {
+                    d.set_content_length(metadata.len());
+                    // last_modified is not available in all platforms.
+                    // it's ok not setting it here.
+                    if let Ok(last_modified) = metadata.modified().map(OffsetDateTime::from) {
+                        d.set_last_modified(last_modified);
+                    }
+                }
+
                 debug!(
-                    "dir object {} got entry, mode: {}, path: {}",
-                    &self.path,
-                    d.mode(),
-                    d.path()
-                );
+                        "dir object {} got entry, mode: {}, path: {}, content length: {:?}, last modified: {:?}, content_md5: {:?}, etag: {:?}",
+                        &self.path,
+                        d.mode(),
+                        d.path(),
+                        d.content_length(),
+                        d.last_modified(),
+                        d.content_md5(),
+                        d.etag()
+                    );
                 Poll::Ready(Some(Ok(d)))
             }
         }