feat(core): Implement list with deleted and versions for cos

.github/services/cos/cos/action.yml

@@ -16,7 +16,7 @@
 # under the License.
 
 name: cos
-description: 'Behavior test for COS. This service is sponsored by @datafuse_labs.'
+description: "Behavior test for COS. This service is sponsored by @datafuse_labs."
 
 runs:
   using: "composite"
@@ -30,3 +30,8 @@ runs:
         OPENDAL_COS_ENDPOINT: op://services/cos/endpoint
         OPENDAL_COS_SECRET_ID: op://services/cos/secret_id
         OPENDAL_COS_SECRET_KEY: op://services/cos/secret_key
+
+    - name: Add extra settings
+      shell: bash
+      run: |
+        echo "OPENDAL_COS_ENABLE_VERSIONING=true" >> $GITHUB_ENV

core/src/services/cos/backend.rs

@@ -29,9 +29,10 @@ use reqsign::TencentCosSigner;
 use super::core::*;
 use super::delete::CosDeleter;
 use super::error::parse_error;
-use super::lister::CosLister;
+use super::lister::{CosLister, CosListers, CosObjectVersionsLister};
 use super::writer::CosWriter;
 use super::writer::CosWriters;
+use crate::raw::oio::PageLister;
 use crate::raw::*;
 use crate::services::CosConfig;
 use crate::*;
@@ -123,6 +124,13 @@ impl CosBuilder {
         self
     }
 
+    /// Set bucket versioning status for this backend
+    pub fn enable_versioning(mut self, enabled: bool) -> Self {
+        self.config.enable_versioning = enabled;
+
+        self
+    }
+
     /// Disable config load so that opendal will not load config from
     /// environment.
     ///
@@ -215,6 +223,7 @@ impl Builder for CosBuilder {
                 bucket: bucket.clone(),
                 root,
                 endpoint: format!("{}://{}.{}", &scheme, &bucket, &endpoint),
+                enable_versioning: self.config.enable_versioning,
                 signer,
                 loader: cred_loader,
                 client,
@@ -232,7 +241,7 @@ pub struct CosBackend {
 impl Access for CosBackend {
     type Reader = HttpBody;
     type Writer = CosWriters;
-    type Lister = oio::PageLister<CosLister>;
+    type Lister = CosListers;
     type Deleter = oio::OneShotDeleter<CosDeleter>;
     type BlockingReader = ();
     type BlockingWriter = ();
@@ -253,15 +262,18 @@ impl Access for CosBackend {
                 stat_has_content_type: true,
                 stat_has_content_encoding: true,
                 stat_has_content_range: true,
+                stat_with_version: self.core.enable_versioning,
                 stat_has_etag: true,
                 stat_has_content_md5: true,
                 stat_has_last_modified: true,
                 stat_has_content_disposition: true,
+                stat_has_version: true,
 
                 read: true,
 
                 read_with_if_match: true,
                 read_with_if_none_match: true,
+                read_with_version: self.core.enable_versioning,
 
                 write: true,
                 write_can_empty: true,
@@ -270,8 +282,8 @@ impl Access for CosBackend {
                 write_with_content_type: true,
                 write_with_cache_control: true,
                 write_with_content_disposition: true,
-                // TODO: set this to false while version has been enabled.
-                write_with_if_not_exists: true,
+                // Cos doesn't support forbid overwrite while version has been enabled.
+                write_with_if_not_exists: !self.core.enable_versioning,
                 // The min multipart size of COS is 1 MiB.
                 //
                 // ref: <https://www.tencentcloud.com/document/product/436/14112>
@@ -286,10 +298,13 @@ impl Access for CosBackend {
                 },
 
                 delete: true,
+                delete_with_version: self.core.enable_versioning,
                 copy: true,
 
                 list: true,
                 list_with_recursive: true,
+                list_with_versions: self.core.enable_versioning,
+                list_with_deleted: self.core.enable_versioning,
                 list_has_content_length: true,
 
                 presign: true,
@@ -311,7 +326,16 @@ impl Access for CosBackend {
         let status = resp.status();
 
         match status {
-            StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+            StatusCode::OK => {
+                let headers = resp.headers();
+                let mut meta = parse_into_metadata(path, headers)?;
+
+                if let Some(v) = parse_header_to_str(headers, "x-cos-version-id")? {
+                    meta.set_version(v);
+                }
+
+                Ok(RpStat::new(meta))
+            }
             _ => Err(parse_error(resp)),
         }
     }
@@ -357,8 +381,22 @@ impl Access for CosBackend {
     }
 
     async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
-        let l = CosLister::new(self.core.clone(), path, args.recursive(), args.limit());
-        Ok((RpList::default(), oio::PageLister::new(l)))
+        let l = if args.versions() || args.deleted() {
+            TwoWays::Two(PageLister::new(CosObjectVersionsLister::new(
+                self.core.clone(),
+                path,
+                args,
+            )))
+        } else {
+            TwoWays::One(PageLister::new(CosLister::new(
+                self.core.clone(),
+                path,
+                args.recursive(),
+                args.limit(),
+            )))
+        };
+
+        Ok((RpList::default(), l))
     }
 
     async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {

core/src/services/cos/config.rs

@@ -35,6 +35,8 @@ pub struct CosConfig {
     pub secret_key: Option<String>,
     /// Bucket of this backend.
     pub bucket: Option<String>,
+    /// is bucket versioning enabled for this bucket
+    pub enable_versioning: bool,
     /// Disable config load so that opendal will not load config from
     pub disable_config_load: bool,
 }

core/src/services/cos/core.rs

@@ -17,6 +17,7 @@
 
 use std::fmt::Debug;
 use std::fmt::Formatter;
+use std::fmt::Write;
 use std::time::Duration;
 
 use bytes::Bytes;
@@ -37,10 +38,15 @@ use serde::Serialize;
 use crate::raw::*;
 use crate::*;
 
+pub mod constants {
+    pub const COS_QUERY_VERSION_ID: &str = "versionId";
+}
+
 pub struct CosCore {
     pub bucket: String,
     pub root: String,
     pub endpoint: String,
+    pub enable_versioning: bool,
 
     pub signer: TencentCosSigner,
     pub loader: TencentCosCredentialLoader,
@@ -125,7 +131,19 @@ impl CosCore {
     ) -> Result<Request<Buffer>> {
         let p = build_abs_path(&self.root, path);
 
-        let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+        let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+        let mut query_args = Vec::new();
+        if let Some(version) = args.version() {
+            query_args.push(format!(
+                "{}={}",
+                constants::COS_QUERY_VERSION_ID,
+                percent_decode_path(version)
+            ))
+        }
+        if !query_args.is_empty() {
+            url.push_str(&format!("?{}", query_args.join("&")));
+        }
 
         let mut req = Request::get(&url);
 
@@ -200,7 +218,19 @@ impl CosCore {
     pub fn cos_head_object_request(&self, path: &str, args: &OpStat) -> Result<Request<Buffer>> {
         let p = build_abs_path(&self.root, path);
 
-        let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+        let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+        let mut query_args = Vec::new();
+        if let Some(version) = args.version() {
+            query_args.push(format!(
+                "{}={}",
+                constants::COS_QUERY_VERSION_ID,
+                percent_decode_path(version)
+            ))
+        }
+        if !query_args.is_empty() {
+            url.push_str(&format!("?{}", query_args.join("&")));
+        }
 
         let mut req = Request::head(&url);
 
@@ -217,10 +247,22 @@ impl CosCore {
         Ok(req)
     }
 
-    pub async fn cos_delete_object(&self, path: &str) -> Result<Response<Buffer>> {
+    pub async fn cos_delete_object(&self, path: &str, args: &OpDelete) -> Result<Response<Buffer>> {
         let p = build_abs_path(&self.root, path);
 
-        let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+        let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+        let mut query_args = Vec::new();
+        if let Some(version) = args.version() {
+            query_args.push(format!(
+                "{}={}",
+                constants::COS_QUERY_VERSION_ID,
+                percent_decode_path(version)
+            ))
+        }
+        if !query_args.is_empty() {
+            url.push_str(&format!("?{}", query_args.join("&")));
+        }
 
         let req = Request::delete(&url);
 
@@ -434,6 +476,50 @@ impl CosCore {
         self.sign(&mut req).await?;
         self.send(req).await
     }
+
+    pub async fn cos_list_object_versions(
+        &self,
+        prefix: &str,
+        delimiter: &str,
+        limit: Option<usize>,
+        key_marker: &str,
+        version_id_marker: &str,
+    ) -> Result<Response<Buffer>> {
+        let p = build_abs_path(&self.root, prefix);
+
+        let mut url = format!("{}?versions", self.endpoint);
+        if !p.is_empty() {
+            write!(url, "&prefix={}", percent_encode_path(p.as_str()))
+                .expect("write into string must succeed");
+        }
+        if !delimiter.is_empty() {
+            write!(url, "&delimiter={}", delimiter).expect("write into string must succeed");
+        }
+
+        if let Some(limit) = limit {
+            write!(url, "&max-keys={}", limit).expect("write into string must succeed");
+        }
+        if !key_marker.is_empty() {
+            write!(url, "&key-marker={}", percent_encode_path(key_marker))
+                .expect("write into string must succeed");
+        }
+        if !version_id_marker.is_empty() {
+            write!(
+                url,
+                "&version-id-marker={}",
+                percent_encode_path(version_id_marker)
+            )
+            .expect("write into string must succeed");
+        }
+
+        let mut req = Request::get(&url)
+            .body(Buffer::new())
+            .map_err(new_request_build_error)?;
+
+        self.sign(&mut req).await?;
+
+        self.send(req).await
+    }
 }
 
 /// Result of CreateMultipartUpload
@@ -511,6 +597,45 @@ pub struct ListObjectsOutputContent {
     pub size: u64,
 }
 
+#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct OutputCommonPrefix {
+    pub prefix: String,
+}
+
+/// Output of ListObjectVersions
+#[derive(Default, Debug, Deserialize)]
+#[serde(default, rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutput {
+    pub is_truncated: Option<bool>,
+    pub next_key_marker: Option<String>,
+    pub next_version_id_marker: Option<String>,
+    pub common_prefixes: Vec<OutputCommonPrefix>,
+    pub version: Vec<ListObjectVersionsOutputVersion>,
+    pub delete_marker: Vec<ListObjectVersionsOutputDeleteMarker>,
+}
+
+#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutputVersion {
+    pub key: String,
+    pub version_id: String,
+    pub is_latest: bool,
+    pub size: u64,
+    pub last_modified: String,
+    #[serde(rename = "ETag")]
+    pub etag: Option<String>,
+}
+
+#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
+#[serde(rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutputDeleteMarker {
+    pub key: String,
+    pub version_id: String,
+    pub is_latest: bool,
+    pub last_modified: String,
+}
+
 #[cfg(test)]
 mod tests {
     use bytes::Buf;

core/src/services/cos/delete.rs

@@ -33,8 +33,8 @@ impl CosDeleter {
 }
 
 impl oio::OneShotDelete for CosDeleter {
-    async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
-        let resp = self.core.cos_delete_object(&path).await?;
+    async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+        let resp = self.core.cos_delete_object(&path, &args).await?;
 
         let status = resp.status();