implement list with deleted in azblob

Author: NikitaMatskevich

core/services/azblob/src/backend.rs

@@ -237,6 +237,15 @@ impl AzblobBuilder {
         self
     }
 
+    /// Set the soft delete feature for this backend.
+    ///
+    /// If enabled, deleted blobs will be retained for the configured retention period
+    /// and can be listed using list_with_deleted.
+    pub fn enable_soft_deletes(mut self, enabled: bool) -> Self {
+        self.config.enable_soft_deletes = enabled;
+        self
+    }
+
     /// from_connection_string will make a builder from connection string
     ///
     /// connection string looks like:
@@ -402,6 +411,12 @@ impl Builder for AzblobBuilder {
 
                             list: true,
                             list_with_recursive: true,
+                            list_has_etag: true,
+                            list_has_content_length: true,
+                            list_has_content_md5: true,
+                            list_has_content_type: true,
+                            list_has_last_modified: true,
+                            list_with_deleted: self.config.enable_soft_deletes,
 
                             presign: self.config.sas_token.is_some(),
                             presign_stat: self.config.sas_token.is_some(),

core/services/azblob/src/core.rs

@@ -599,6 +599,7 @@ impl AzblobCore {
         next_marker: &str,
         delimiter: &str,
         limit: Option<usize>,
+        include_deleted: bool,
     ) -> Result<Response<Buffer>> {
         let p = build_abs_path(&self.root, path);
         let mut url = QueryPairsWriter::new(&format!("{}/{}", self.endpoint, self.container))
@@ -618,6 +619,10 @@ impl AzblobCore {
             url = url.push("marker", next_marker);
         }
 
+        if include_deleted {
+            url = url.push("include", "deleted");
+        }
+
         let mut req = Request::get(url.finish())
             .extension(Operation::List)
             .body(Buffer::new())
@@ -685,6 +690,8 @@ pub struct BlobPrefix {
 pub struct Blob {
     pub properties: Properties,
     pub name: String,
+    #[serde(rename = "Deleted")]
+    pub deleted: Option<bool>,
 }
 
 #[derive(Default, Debug, Deserialize)]

core/services/azblob/src/lister.rs

@@ -32,17 +32,25 @@ pub struct AzblobLister {
     path: String,
     delimiter: &'static str,
     limit: Option<usize>,
+    deleted: bool,
 }
 
 impl AzblobLister {
-    pub fn new(core: Arc<AzblobCore>, path: String, recursive: bool, limit: Option<usize>) -> Self {
+    pub fn new(
+        core: Arc<AzblobCore>,
+        path: String,
+        recursive: bool,
+        limit: Option<usize>,
+        deleted: bool,
+    ) -> Self {
         let delimiter = if recursive { "" } else { "/" };
 
         Self {
             core,
             path,
             delimiter,
             limit,
+            deleted,
         }
     }
 }
@@ -51,7 +59,13 @@ impl oio::PageList for AzblobLister {
     async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
         let resp = self
             .core
-            .azblob_list_blobs(&self.path, &ctx.token, self.delimiter, self.limit)
+            .azblob_list_blobs(
+                &self.path,
+                &ctx.token,
+                self.delimiter,
+                self.limit,
+                self.deleted,
+            )
             .await?;
 
         if resp.status() != http::StatusCode::OK {
@@ -88,7 +102,7 @@ impl oio::PageList for AzblobLister {
                 path = "/".to_string();
             }
 
-            let meta = Metadata::new(EntryMode::from_path(&path))
+            let mut meta = Metadata::new(EntryMode::from_path(&path))
                 // Keep fit with ETag header.
                 .with_etag(format!("\"{}\"", object.properties.etag.as_str()))
                 .with_content_length(object.properties.content_length)
@@ -98,6 +112,11 @@ impl oio::PageList for AzblobLister {
                     object.properties.last_modified.as_str(),
                 )?);
 
+            // Mark as deleted if this is a delete marker
+            if object.deleted.unwrap_or(false) {
+                meta = meta.with_is_deleted(true);
+            }
+
             let de = oio::Entry::with(path, meta);
             ctx.entries.push_back(de);
         }

core/src/services/azblob/config.rs

@@ -0,0 +1,90 @@
+// 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 std::fmt::Debug;
+use std::fmt::Formatter;
+
+use serde::Deserialize;
+use serde::Serialize;
+
+/// Azure Storage Blob services support.
+#[derive(Default, Serialize, Deserialize, Clone, PartialEq, Eq)]
+pub struct AzblobConfig {
+    /// The root of Azblob service backend.
+    ///
+    /// All operations will happen under this root.
+    pub root: Option<String>,
+
+    /// The container name of Azblob service backend.
+    pub container: String,
+
+    /// The endpoint of Azblob service backend.
+    ///
+    /// Endpoint must be full uri, e.g.
+    ///
+    /// - Azblob: `https://accountname.blob.core.windows.net`
+    /// - Azurite: `http://127.0.0.1:10000/devstoreaccount1`
+    pub endpoint: Option<String>,
+
+    /// The account name of Azblob service backend.
+    pub account_name: Option<String>,
+
+    /// The account key of Azblob service backend.
+    pub account_key: Option<String>,
+
+    /// The encryption key of Azblob service backend.
+    pub encryption_key: Option<String>,
+
+    /// The encryption key sha256 of Azblob service backend.
+    pub encryption_key_sha256: Option<String>,
+
+    /// The encryption algorithm of Azblob service backend.
+    pub encryption_algorithm: Option<String>,
+
+    /// The sas token of Azblob service backend.
+    pub sas_token: Option<String>,
+
+    /// The maximum batch operations of Azblob service backend.
+    pub batch_max_operations: Option<usize>,
+
+    /// Enable soft deletes for this storage account.
+    #[serde(default)]
+    pub enable_soft_deletes: bool,
+}
+
+impl Debug for AzblobConfig {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let mut ds = f.debug_struct("AzblobConfig");
+
+        ds.field("root", &self.root);
+        ds.field("container", &self.container);
+        ds.field("endpoint", &self.endpoint);
+
+        if self.account_name.is_some() {
+            ds.field("account_name", &"<redacted>");
+        }
+        if self.account_key.is_some() {
+            ds.field("account_key", &"<redacted>");
+        }
+        if self.sas_token.is_some() {
+            ds.field("sas_token", &"<redacted>");
+        }
+        ds.field("enable_soft_deletes", &self.enable_soft_deletes);
+
+        ds.finish()
+    }
+}