You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2024/12/05 11:43:30 UTC
(opendal) 02/09: Save work
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch deleter
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 5eaeb626cc82c566bfa7cd26ae85101d2664bb6f
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Nov 22 18:08:11 2024 +0800
Save work
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/docs/rfcs/3911_deleter_api.md | 2 +-
core/src/layers/async_backtrace.rs | 50 +++-
core/src/layers/await_tree.rs | 44 ++-
core/src/layers/blocking.rs | 28 +-
core/src/layers/chaos.rs | 11 +
core/src/layers/complete.rs | 25 +-
core/src/layers/concurrent_limit.rs | 56 ++--
core/src/layers/dtrace.rs | 41 +--
core/src/layers/error_context.rs | 120 +++++---
core/src/layers/fastrace.rs | 92 +++++-
core/src/layers/immutable_index.rs | 23 +-
core/src/layers/logging.rs | 331 +++++++++++++++------
core/src/layers/mime_guess.rs | 41 ++-
core/src/layers/observe/metrics.rs | 214 ++++++++++---
core/src/layers/oteltrace.rs | 29 +-
core/src/layers/retry.rs | 155 +++++-----
core/src/layers/throttle.rs | 15 +-
core/src/layers/timeout.rs | 28 +-
core/src/layers/tracing.rs | 48 ++-
core/src/layers/type_eraser.rs | 25 +-
core/src/raw/accessor.rs | 85 ++----
core/src/raw/adapters/kv/backend.rs | 112 ++++---
core/src/raw/adapters/typed_kv/backend.rs | 109 ++++---
core/src/raw/http_util/header.rs | 25 ++
core/src/raw/http_util/mod.rs | 1 +
core/src/raw/layer.rs | 66 ++--
core/src/raw/oio/delete/api.rs | 66 ++--
core/src/raw/oio/delete/batch_delete.rs | 125 ++++++++
core/src/raw/oio/delete/mod.rs | 16 +
core/src/raw/oio/delete/one_shot_delete.rs | 104 +++++++
core/src/raw/operation.rs | 16 +-
core/src/raw/ops.rs | 60 +---
core/src/raw/rps.rs | 34 ---
core/src/services/aliyun_drive/backend.rs | 20 +-
core/src/services/aliyun_drive/delete.rs | 48 +++
core/src/services/aliyun_drive/mod.rs | 2 +
core/src/services/alluxio/backend.rs | 9 +-
.../services/{azblob/mod.rs => alluxio/delete.rs} | 32 +-
core/src/services/alluxio/mod.rs | 2 +
core/src/services/azblob/backend.rs | 94 +-----
core/src/services/azblob/delete.rs | 103 +++++++
core/src/services/azblob/mod.rs | 2 +
core/src/services/azdls/backend.rs | 17 +-
.../services/{azblob/mod.rs => azdls/delete.rs} | 41 ++-
core/src/services/azdls/mod.rs | 2 +
core/src/services/dropbox/backend.rs | 51 ----
core/src/services/dropbox/core.rs | 122 --------
core/src/services/lakefs/backend.rs | 2 +-
core/src/services/oss/backend.rs | 2 +-
core/src/services/s3/backend.rs | 70 +----
core/src/services/s3/delete.rs | 111 +++++++
core/src/services/s3/mod.rs | 2 +
core/src/types/capability.rs | 2 +
core/src/types/delete.rs | 197 ++++++++++++
core/src/types/mod.rs | 2 +
core/src/types/operator/operator.rs | 86 ++----
core/src/types/operator/operator_futures.rs | 5 +
57 files changed, 2071 insertions(+), 1150 deletions(-)
diff --git a/core/src/docs/rfcs/3911_deleter_api.md b/core/src/docs/rfcs/3911_deleter_api.md
index dfdc7a73c..f7ab31461 100644
--- a/core/src/docs/rfcs/3911_deleter_api.md
+++ b/core/src/docs/rfcs/3911_deleter_api.md
@@ -128,7 +128,7 @@ And the `delete` API will be changed to return a `oio::Delete` instead:
```diff
trait Accessor {
-- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete>;
+- async fn delete(&self) -> Result<(RpDelete, Self::Deleter)>;
+ async fn delete(&self, args: OpDelete) -> Result<(RpDelete, Self::Deleter)>;
}
```
diff --git a/core/src/layers/async_backtrace.rs b/core/src/layers/async_backtrace.rs
index 65a5c3305..91d03d1c8 100644
--- a/core/src/layers/async_backtrace.rs
+++ b/core/src/layers/async_backtrace.rs
@@ -17,6 +17,7 @@
use crate::raw::*;
use crate::*;
+use std::future::Future;
/// Add Efficient, logical 'stack' traces of async functions for the underlying services.
///
@@ -65,6 +66,8 @@ impl<A: Access> LayeredAccess for AsyncBacktraceAccessor<A> {
type BlockingWriter = AsyncBacktraceWrapper<A::BlockingWriter>;
type Lister = AsyncBacktraceWrapper<A::Lister>;
type BlockingLister = AsyncBacktraceWrapper<A::BlockingLister>;
+ type Deleter = AsyncBacktraceWrapper<A::Lister>;
+ type BlockingDeleter = AsyncBacktraceWrapper<A::BlockingLister>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -102,8 +105,11 @@ impl<A: Access> LayeredAccess for AsyncBacktraceAccessor<A> {
}
#[async_backtrace::framed]
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.delete(path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner
+ .delete()
+ .await
+ .map(|(rp, r)| (rp, AsyncBacktraceWrapper::new(r)))
}
#[async_backtrace::framed]
@@ -114,11 +120,6 @@ impl<A: Access> LayeredAccess for AsyncBacktraceAccessor<A> {
.map(|(rp, r)| (rp, AsyncBacktraceWrapper::new(r)))
}
- #[async_backtrace::framed]
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.inner.batch(args).await
- }
-
#[async_backtrace::framed]
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
self.inner.presign(path, args).await
@@ -141,6 +142,12 @@ impl<A: Access> LayeredAccess for AsyncBacktraceAccessor<A> {
.blocking_list(path, args)
.map(|(rp, r)| (rp, AsyncBacktraceWrapper::new(r)))
}
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner
+ .blocking_delete()
+ .map(|(rp, r)| (rp, AsyncBacktraceWrapper::new(r)))
+ }
}
pub struct AsyncBacktraceWrapper<R> {
@@ -173,13 +180,13 @@ impl<R: oio::Write> oio::Write for AsyncBacktraceWrapper<R> {
}
#[async_backtrace::framed]
- async fn abort(&mut self) -> Result<()> {
- self.inner.abort().await
+ async fn close(&mut self) -> Result<()> {
+ self.inner.close().await
}
#[async_backtrace::framed]
- async fn close(&mut self) -> Result<()> {
- self.inner.close().await
+ async fn abort(&mut self) -> Result<()> {
+ self.inner.abort().await
}
}
@@ -205,3 +212,24 @@ impl<R: oio::BlockingList> oio::BlockingList for AsyncBacktraceWrapper<R> {
self.inner.next()
}
}
+
+impl<R: oio::Delete> oio::Delete for AsyncBacktraceWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ #[async_backtrace::framed]
+ async fn flush(&mut self) -> Result<usize> {
+ self.inner.flush().await
+ }
+}
+
+impl<R: oio::BlockingDelete> oio::BlockingDelete for AsyncBacktraceWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ self.inner.flush()
+ }
+}
diff --git a/core/src/layers/await_tree.rs b/core/src/layers/await_tree.rs
index 3c1b17d64..ae5b7ac71 100644
--- a/core/src/layers/await_tree.rs
+++ b/core/src/layers/await_tree.rs
@@ -76,6 +76,8 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
type BlockingWriter = AwaitTreeWrapper<A::BlockingWriter>;
type Lister = AwaitTreeWrapper<A::Lister>;
type BlockingLister = AwaitTreeWrapper<A::BlockingLister>;
+ type Deleter = AwaitTreeWrapper<A::Deleter>;
+ type BlockingDeleter = AwaitTreeWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -118,11 +120,12 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
.await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner
- .delete(path, args)
+ .delete()
.instrument_await(format!("opendal::{}", Operation::Delete))
.await
+ .map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -140,13 +143,6 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
.await
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.inner
- .batch(args)
- .instrument_await(format!("opendal::{}", Operation::Batch))
- .await
- }
-
fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
self.inner
.blocking_read(path, args)
@@ -164,6 +160,13 @@ impl<A: Access> LayeredAccess for AwaitTreeAccessor<A> {
.blocking_list(path, args)
.map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
}
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner
+ .blocking_delete()
+ .instrument_await(format!("opendal::{}", Operation::BlockingDelete))
+ .map(|(rp, r)| (rp, AwaitTreeWrapper::new(r)))
+ }
}
pub struct AwaitTreeWrapper<R> {
@@ -235,3 +238,26 @@ impl<R: oio::BlockingList> oio::BlockingList for AwaitTreeWrapper<R> {
self.inner.next()
}
}
+
+impl<R: oio::Delete> oio::Delete for AwaitTreeWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ self.inner
+ .flush()
+ .instrument_await(format!("opendal::{}", Operation::DeleterFlush))
+ .await
+ }
+}
+
+impl<R: oio::BlockingDelete> oio::BlockingDelete for AwaitTreeWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ self.inner.flush()
+ }
+}
diff --git a/core/src/layers/blocking.rs b/core/src/layers/blocking.rs
index a68538ec6..512159a23 100644
--- a/core/src/layers/blocking.rs
+++ b/core/src/layers/blocking.rs
@@ -168,6 +168,8 @@ impl<A: Access> LayeredAccess for BlockingAccessor<A> {
type BlockingWriter = BlockingWrapper<A::Writer>;
type Lister = A::Lister;
type BlockingLister = BlockingWrapper<A::Lister>;
+ type Deleter = A::Deleter;
+ type BlockingDeleter = BlockingWrapper<A::Deleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -203,8 +205,8 @@ impl<A: Access> LayeredAccess for BlockingAccessor<A> {
self.inner.stat(path, args).await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.delete(path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -215,10 +217,6 @@ impl<A: Access> LayeredAccess for BlockingAccessor<A> {
self.inner.presign(path, args).await
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.inner.batch(args).await
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.handle.block_on(self.inner.create_dir(path, args))
}
@@ -252,8 +250,12 @@ impl<A: Access> LayeredAccess for BlockingAccessor<A> {
self.handle.block_on(self.inner.stat(path, args))
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.handle.block_on(self.inner.delete(path, args))
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.handle.block_on(async {
+ let (rp, writer) = self.inner.delete().await?;
+ let blocking_deleter = Self::BlockingDeleter::new(self.handle.clone(), writer);
+ Ok((rp, blocking_deleter))
+ })
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -298,6 +300,16 @@ impl<I: oio::List> oio::BlockingList for BlockingWrapper<I> {
}
}
+impl<I: oio::Delete> oio::BlockingDelete for BlockingWrapper<I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ self.handle.block_on(self.inner.flush())
+ }
+}
+
#[cfg(test)]
mod tests {
use once_cell::sync::Lazy;
diff --git a/core/src/layers/chaos.rs b/core/src/layers/chaos.rs
index 2432d21e0..0c39a7c68 100644
--- a/core/src/layers/chaos.rs
+++ b/core/src/layers/chaos.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::sync::Arc;
use std::sync::Mutex;
@@ -105,6 +106,8 @@ impl<A: Access> LayeredAccess for ChaosAccessor<A> {
type BlockingWriter = A::BlockingWriter;
type Lister = A::Lister;
type BlockingLister = A::BlockingLister;
+ type Deleter = A::Deleter;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -138,6 +141,14 @@ impl<A: Access> LayeredAccess for ChaosAccessor<A> {
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.inner.blocking_list(path, args)
}
+
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
}
/// ChaosReader will inject error into read operations.
diff --git a/core/src/layers/complete.rs b/core/src/layers/complete.rs
index 2d1787638..1a5e2cc54 100644
--- a/core/src/layers/complete.rs
+++ b/core/src/layers/complete.rs
@@ -361,6 +361,8 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
type BlockingWriter = CompleteWriter<A::BlockingWriter>;
type Lister = CompleteLister<A, A::Lister>;
type BlockingLister = CompleteLister<A, A::BlockingLister>;
+ type Deleter = A::Deleter;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -435,13 +437,13 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
self.complete_stat(path, args).await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
let capability = self.info.full_capability();
if !capability.delete {
return Err(self.new_unsupported_error(Operation::Delete));
}
- self.inner().delete(path, args).await
+ self.inner().delete().await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -453,15 +455,6 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
self.complete_list(path, args).await
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let capability = self.info.full_capability();
- if !capability.batch {
- return Err(self.new_unsupported_error(Operation::Batch));
- }
-
- self.inner().batch(args).await
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
let capability = self.info.full_capability();
if !capability.presign {
@@ -530,13 +523,13 @@ impl<A: Access> LayeredAccess for CompleteAccessor<A> {
self.complete_blocking_stat(path, args)
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
let capability = self.info.full_capability();
if !capability.delete || !capability.blocking {
return Err(self.new_unsupported_error(Operation::BlockingDelete));
}
- self.inner().blocking_delete(path, args)
+ self.inner().blocking_delete()
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -713,9 +706,11 @@ mod tests {
type Reader = oio::Reader;
type Writer = oio::Writer;
type Lister = oio::Lister;
+ type Deleter = oio::Deleter;
type BlockingReader = oio::BlockingReader;
type BlockingWriter = oio::BlockingWriter;
type BlockingLister = oio::BlockingLister;
+ type BlockingDeleter = oio::BlockingDeleter;
fn info(&self) -> Arc<AccessorInfo> {
let mut info = AccessorInfo::default();
@@ -740,10 +735,6 @@ mod tests {
Ok((RpWrite::new(), Box::new(())))
}
- async fn delete(&self, _: &str, _: OpDelete) -> Result<RpDelete> {
- Ok(RpDelete {})
- }
-
async fn list(&self, _: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
Ok((RpList {}, Box::new(())))
}
diff --git a/core/src/layers/concurrent_limit.rs b/core/src/layers/concurrent_limit.rs
index bd1766644..19d517304 100644
--- a/core/src/layers/concurrent_limit.rs
+++ b/core/src/layers/concurrent_limit.rs
@@ -16,6 +16,7 @@
// under the License.
use std::fmt::Debug;
+use std::future::Future;
use std::sync::Arc;
use tokio::sync::OwnedSemaphorePermit;
@@ -84,6 +85,8 @@ impl<A: Access> LayeredAccess for ConcurrentLimitAccessor<A> {
type BlockingWriter = ConcurrentLimitWrapper<A::BlockingWriter>;
type Lister = ConcurrentLimitWrapper<A::Lister>;
type BlockingLister = ConcurrentLimitWrapper<A::BlockingLister>;
+ type Deleter = ConcurrentLimitWrapper<A::Deleter>;
+ type BlockingDeleter = ConcurrentLimitWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -137,14 +140,18 @@ impl<A: Access> LayeredAccess for ConcurrentLimitAccessor<A> {
self.inner.stat(path, args).await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let _permit = self
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ let permit = self
.semaphore
- .acquire()
+ .clone()
+ .acquire_owned()
.await
.expect("semaphore must be valid");
- self.inner.delete(path, args).await
+ self.inner
+ .delete()
+ .await
+ .map(|(rp, w)| (rp, ConcurrentLimitWrapper::new(w, permit)))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -161,16 +168,6 @@ impl<A: Access> LayeredAccess for ConcurrentLimitAccessor<A> {
.map(|(rp, s)| (rp, ConcurrentLimitWrapper::new(s, permit)))
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let _permit = self
- .semaphore
- .acquire()
- .await
- .expect("semaphore must be valid");
-
- self.inner.batch(args).await
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
let _permit = self
.semaphore
@@ -213,13 +210,16 @@ impl<A: Access> LayeredAccess for ConcurrentLimitAccessor<A> {
self.inner.blocking_stat(path, args)
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let _permit = self
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ let permit = self
.semaphore
- .try_acquire()
+ .clone()
+ .try_acquire_owned()
.expect("semaphore must be valid");
- self.inner.blocking_delete(path, args)
+ self.inner
+ .blocking_delete()
+ .map(|(rp, w)| (rp, ConcurrentLimitWrapper::new(w, permit)))
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -298,3 +298,23 @@ impl<R: oio::BlockingList> oio::BlockingList for ConcurrentLimitWrapper<R> {
self.inner.next()
}
}
+
+impl<R: oio::Delete> oio::Delete for ConcurrentLimitWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ self.inner.flush().await
+ }
+}
+
+impl<R: oio::BlockingDelete> oio::BlockingDelete for ConcurrentLimitWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ self.inner.flush()
+ }
+}
diff --git a/core/src/layers/dtrace.rs b/core/src/layers/dtrace.rs
index 7cfa8ac55..65533c008 100644
--- a/core/src/layers/dtrace.rs
+++ b/core/src/layers/dtrace.rs
@@ -68,34 +68,21 @@ use crate::*;
/// 1. reader_read_start, arguments: path
/// 2. reader_read_ok, arguments: path, length
/// 3. reader_read_error, arguments: path
-/// 4. reader_seek_start, arguments: path
-/// 5. reader_seek_ok, arguments: path, offset
-/// 6. reader_seek_error, arguments: path
-/// 7. reader_next_start, arguments: path
-/// 8. reader_next_ok, arguments: path, length
-/// 9. reader_next_error, arguments: path
-/// 10. reader_next_end, arguments: path
///
/// ### For BlockingReader
///
/// 1. blocking_reader_read_start, arguments: path
/// 2. blocking_reader_read_ok, arguments: path, length
/// 3. blocking_reader_read_error, arguments: path
-/// 4. blocking_reader_seek_start, arguments: path
-/// 5. blocking_reader_seek_ok, arguments: path, offset
-/// 6. blocking_reader_seek_error, arguments: path
-/// 7. blocking_reader_next_start, arguments: path
-/// 8. blocking_reader_next_ok, arguments: path, length
-/// 9. blocking_reader_next_error, arguments: path
///
/// ### For Writer
///
/// 1. writer_write_start, arguments: path
/// 2. writer_write_ok, arguments: path, length
/// 3. writer_write_error, arguments: path
-/// 4. writer_poll_abort_start, arguments: path
-/// 5. writer_poll_abort_ok, arguments: path
-/// 6. writer_poll_abort_error, arguments: path
+/// 4. writer_abort_start, arguments: path
+/// 5. writer_abort_ok, arguments: path
+/// 6. writer_abort_error, arguments: path
/// 7. writer_close_start, arguments: path
/// 8. writer_close_ok, arguments: path
/// 9. writer_close_error, arguments: path
@@ -192,6 +179,8 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
type BlockingWriter = DtraceLayerWrapper<A::BlockingWriter>;
type Lister = A::Lister;
type BlockingLister = A::BlockingLister;
+ type Deleter = A::Deleter;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -238,12 +227,8 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
result
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, delete_start, c_path.as_ptr());
- let result = self.inner.delete(path, args).await;
- probe_lazy!(opendal, delete_end, c_path.as_ptr());
- result
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -254,10 +239,6 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
result
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.inner.batch(args).await
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
let c_path = CString::new(path).unwrap();
probe_lazy!(opendal, presign_start, c_path.as_ptr());
@@ -304,12 +285,8 @@ impl<A: Access> LayeredAccess for DTraceAccessor<A> {
result
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let c_path = CString::new(path).unwrap();
- probe_lazy!(opendal, blocking_delete_start, c_path.as_ptr());
- let result = self.inner.blocking_delete(path, args);
- probe_lazy!(opendal, blocking_delete_end, c_path.as_ptr());
- result
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.blocking_delete()
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index a160a6377..a27de9718 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::future::Future;
use std::sync::Arc;
use crate::raw::*;
@@ -39,6 +40,7 @@ use crate::*;
/// - `size`: The size of the current write operation.
/// - `written`: The already written size in given writer.
/// - `listed`: The already listed size in given lister.
+/// - `deleted`: The already deleted size in given deleter.
pub struct ErrorContextLayer;
impl<A: Access> Layer<A> for ErrorContextLayer {
@@ -70,6 +72,8 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
type BlockingWriter = ErrorContextWrapper<A::BlockingWriter>;
type Lister = ErrorContextWrapper<A::Lister>;
type BlockingLister = ErrorContextWrapper<A::BlockingLister>;
+ type Deleter = ErrorContextWrapper<A::Deleter>;
+ type BlockingDeleter = ErrorContextWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -150,54 +154,36 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
})
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.delete(path, args).await.map_err(|err| {
- err.with_operation(Operation::Delete)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- })
- }
-
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
self.inner
- .list(path, args)
+ .delete()
.await
- .map(|(rp, p)| {
+ .map(|(rp, w)| {
(
rp,
- ErrorContextWrapper::new(self.info.scheme(), path.to_string(), p),
+ ErrorContextWrapper::new(self.info.scheme(), "".to_string(), w),
)
})
.map_err(|err| {
- err.with_operation(Operation::List)
+ err.with_operation(Operation::Delete)
.with_context("service", self.info.scheme())
- .with_context("path", path)
})
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner
- .batch(args)
+ .list(path, args)
.await
- .map(|v| {
- let res = v
- .into_results()
- .into_iter()
- .map(|(path, res)| {
- let res = res.map_err(|err| {
- err.with_operation(Operation::Delete)
- .with_context("service", self.info.scheme())
- .with_context("path", &path)
- });
- (path, res)
- })
- .collect();
-
- RpBatch::new(res)
+ .map(|(rp, p)| {
+ (
+ rp,
+ ErrorContextWrapper::new(self.info.scheme(), path.to_string(), p),
+ )
})
.map_err(|err| {
- err.with_operation(Operation::Batch)
+ err.with_operation(Operation::List)
.with_context("service", self.info.scheme())
+ .with_context("path", path)
})
}
@@ -278,12 +264,19 @@ impl<A: Access> LayeredAccess for ErrorContextAccessor<A> {
})
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.blocking_delete(path, args).map_err(|err| {
- err.with_operation(Operation::BlockingDelete)
- .with_context("service", self.info.scheme())
- .with_context("path", path)
- })
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner
+ .blocking_delete()
+ .map(|(rp, w)| {
+ (
+ rp,
+ ErrorContextWrapper::new(self.info.scheme(), "".to_string(), w),
+ )
+ })
+ .map_err(|err| {
+ err.with_operation(Operation::BlockingDelete)
+ .with_context("service", self.info.scheme())
+ })
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -463,3 +456,54 @@ impl<T: oio::BlockingList> oio::BlockingList for ErrorContextWrapper<T> {
})
}
}
+
+impl<T: oio::Delete> oio::Delete for ErrorContextWrapper<T> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args).map_err(|err| {
+ err.with_operation(Operation::DeleterDelete)
+ .with_context("service", self.scheme)
+ .with_context("path", path)
+ .with_context("deleted", self.processed.to_string())
+ })
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ self.inner
+ .flush()
+ .await
+ .map(|n| {
+ self.processed += n as u64;
+ n
+ })
+ .map_err(|err| {
+ err.with_operation(Operation::DeleterFlush)
+ .with_context("service", self.scheme)
+ .with_context("deleted", self.processed.to_string())
+ })
+ }
+}
+
+impl<T: oio::BlockingDelete> oio::BlockingDelete for ErrorContextWrapper<T> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args).map_err(|err| {
+ err.with_operation(Operation::DeleterDelete)
+ .with_context("service", self.scheme)
+ .with_context("path", path)
+ .with_context("deleted", self.processed.to_string())
+ })
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ self.inner
+ .flush()
+ .map(|n| {
+ self.processed += n as u64;
+ n
+ })
+ .map_err(|err| {
+ err.with_operation(Operation::DeleterFlush)
+ .with_context("service", self.scheme)
+ .with_context("deleted", self.processed.to_string())
+ })
+ }
+}
diff --git a/core/src/layers/fastrace.rs b/core/src/layers/fastrace.rs
index 51153d616..b392a91c8 100644
--- a/core/src/layers/fastrace.rs
+++ b/core/src/layers/fastrace.rs
@@ -126,6 +126,8 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
type BlockingWriter = FastraceWrapper<A::BlockingWriter>;
type Lister = FastraceWrapper<A::Lister>;
type BlockingLister = FastraceWrapper<A::BlockingLister>;
+ type Deleter = FastraceWrapper<A::Deleter>;
+ type BlockingDeleter = FastraceWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -136,7 +138,7 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.info()
}
- #[trace(name = "create", enter_on_poll = true)]
+ #[trace(enter_on_poll = true)]
async fn create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.inner.create_dir(path, args).await
}
@@ -146,7 +148,10 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.read(path, args).await.map(|(rp, r)| {
(
rp,
- FastraceWrapper::new(Span::enter_with_local_parent("ReadOperation"), r),
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::Read.into_static()),
+ r,
+ ),
)
})
}
@@ -156,7 +161,10 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.write(path, args).await.map(|(rp, r)| {
(
rp,
- FastraceWrapper::new(Span::enter_with_local_parent("WriteOperation"), r),
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::Write.into_static()),
+ r,
+ ),
)
})
}
@@ -177,8 +185,16 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
}
#[trace(enter_on_poll = true)]
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.delete(path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await.map(|(rp, r)| {
+ (
+ rp,
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::Delete.into_static()),
+ r,
+ ),
+ )
+ })
}
#[trace(enter_on_poll = true)]
@@ -186,7 +202,10 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.list(path, args).await.map(|(rp, s)| {
(
rp,
- FastraceWrapper::new(Span::enter_with_local_parent("ListOperation"), s),
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::List.into_static()),
+ s,
+ ),
)
})
}
@@ -196,11 +215,6 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.presign(path, args).await
}
- #[trace(enter_on_poll = true)]
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.inner.batch(args).await
- }
-
#[trace]
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.inner.blocking_create_dir(path, args)
@@ -211,7 +225,10 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.blocking_read(path, args).map(|(rp, r)| {
(
rp,
- FastraceWrapper::new(Span::enter_with_local_parent("ReadOperation"), r),
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::BlockingRead.into_static()),
+ r,
+ ),
)
})
}
@@ -221,7 +238,10 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.blocking_write(path, args).map(|(rp, r)| {
(
rp,
- FastraceWrapper::new(Span::enter_with_local_parent("WriteOperation"), r),
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::BlockingWrite.into_static()),
+ r,
+ ),
)
})
}
@@ -242,8 +262,16 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
}
#[trace]
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.blocking_delete(path, args)
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.blocking_delete().map(|(rp, r)| {
+ (
+ rp,
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::BlockingDelete.into_static()),
+ r,
+ ),
+ )
+ })
}
#[trace]
@@ -251,7 +279,10 @@ impl<A: Access> LayeredAccess for FastraceAccessor<A> {
self.inner.blocking_list(path, args).map(|(rp, it)| {
(
rp,
- FastraceWrapper::new(Span::enter_with_local_parent("PageOperation"), it),
+ FastraceWrapper::new(
+ Span::enter_with_local_parent(Operation::BlockingList.into_static()),
+ it,
+ ),
)
})
}
@@ -333,3 +364,32 @@ impl<R: oio::BlockingList> oio::BlockingList for FastraceWrapper<R> {
self.inner.next()
}
}
+
+impl<R: oio::Delete> oio::Delete for FastraceWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let _g = self.span.set_local_parent();
+ let _span = LocalSpan::enter_with_local_parent(Operation::DeleterDelete.into_static());
+ self.inner.delete(path, args)
+ }
+
+ #[trace(enter_on_poll = true)]
+ async fn flush(&mut self) -> Result<usize> {
+ self.inner.flush().await
+ }
+}
+
+impl<R: oio::BlockingDelete> oio::BlockingDelete for FastraceWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let _g = self.span.set_local_parent();
+ let _span =
+ LocalSpan::enter_with_local_parent(Operation::BlockingDeleterDelete.into_static());
+ self.inner.delete(path, args)
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ let _g = self.span.set_local_parent();
+ let _span =
+ LocalSpan::enter_with_local_parent(Operation::BlockingDeleterFlush.into_static());
+ self.inner.flush()
+ }
+}
diff --git a/core/src/layers/immutable_index.rs b/core/src/layers/immutable_index.rs
index 4a54adb81..1d8c1c9d2 100644
--- a/core/src/layers/immutable_index.rs
+++ b/core/src/layers/immutable_index.rs
@@ -17,6 +17,7 @@
use std::collections::HashSet;
use std::fmt::Debug;
+use std::future::Future;
use std::sync::Arc;
use std::vec::IntoIter;
@@ -138,11 +139,13 @@ impl<A: Access> ImmutableIndexAccessor<A> {
impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
type Inner = A;
type Reader = A::Reader;
- type BlockingReader = A::BlockingReader;
type Writer = A::Writer;
- type BlockingWriter = A::BlockingWriter;
type Lister = ImmutableDir;
+ type Deleter = A::Deleter;
+ type BlockingReader = A::BlockingReader;
+ type BlockingWriter = A::BlockingWriter;
type BlockingLister = ImmutableDir;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -163,6 +166,10 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
self.inner.read(path, args).await
}
+ async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
+ self.inner.write(path, args).await
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
let mut path = path;
if path == "/" {
@@ -178,12 +185,12 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
Ok((RpList::default(), ImmutableDir::new(idx)))
}
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
- self.inner.blocking_read(path, args)
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
}
- async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
- self.inner.write(path, args).await
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ self.inner.blocking_read(path, args)
}
fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
@@ -204,6 +211,10 @@ impl<A: Access> LayeredAccess for ImmutableIndexAccessor<A> {
Ok((RpList::default(), ImmutableDir::new(idx)))
}
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
}
pub struct ImmutableDir {
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index ec929e50d..a631d9919 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Display;
+use std::future::Future;
use std::sync::Arc;
use log::log;
@@ -261,6 +262,8 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
type BlockingWriter = LoggingWriter<A::BlockingWriter, I>;
type Lister = LoggingLister<A::Lister, I>;
type BlockingLister = LoggingLister<A::BlockingLister, I>;
+ type Deleter = LoggingDeleter<A::Deleter, I>;
+ type BlockingDeleter = LoggingDeleter<A::BlockingDeleter, I>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -486,36 +489,22 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
})
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[("path", path)],
- "started",
- None,
- );
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.logger
+ .log(&self.info, Operation::Delete, &[], "started", None);
self.inner
- .delete(path, args.clone())
+ .delete()
.await
- .map(|v| {
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[("path", path)],
- "finished",
- None,
- );
- v
+ .map(|(rp, d)| {
+ self.logger
+ .log(&self.info, Operation::Delete, &[], "finished", None);
+ let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
+ (rp, d)
})
.map_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Delete,
- &[("path", path)],
- "failed",
- Some(&err),
- );
+ self.logger
+ .log(&self.info, Operation::Delete, &[], "failed", Some(&err));
err
})
}
@@ -589,47 +578,6 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
})
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let (op, count) = (args.operation()[0].1.operation(), args.operation().len());
-
- self.logger.log(
- &self.info,
- Operation::Batch,
- &[("op", op.into_static()), ("count", &count.to_string())],
- "started",
- None,
- );
-
- self.inner
- .batch(args)
- .await
- .map(|v| {
- self.logger.log(
- &self.info,
- Operation::Batch,
- &[("op", op.into_static()), ("count", &count.to_string())],
- &format!(
- "finished: {}, succeed: {}, failed: {}",
- v.results().len(),
- v.results().iter().filter(|(_, v)| v.is_ok()).count(),
- v.results().iter().filter(|(_, v)| v.is_err()).count(),
- ),
- None,
- );
- v
- })
- .map_err(|err| {
- self.logger.log(
- &self.info,
- Operation::Batch,
- &[("op", op.into_static()), ("count", &count.to_string())],
- "failed",
- Some(&err),
- );
- err
- })
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.logger.log(
&self.info,
@@ -830,32 +778,23 @@ impl<A: Access, I: LoggingInterceptor> LayeredAccess for LoggingAccessor<A, I> {
})
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.logger.log(
- &self.info,
- Operation::BlockingDelete,
- &[("path", path)],
- "started",
- None,
- );
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.logger
+ .log(&self.info, Operation::BlockingDelete, &[], "started", None);
self.inner
- .blocking_delete(path, args)
- .map(|v| {
- self.logger.log(
- &self.info,
- Operation::BlockingDelete,
- &[("path", path)],
- "finished",
- None,
- );
- v
+ .blocking_delete()
+ .map(|(rp, d)| {
+ self.logger
+ .log(&self.info, Operation::BlockingDelete, &[], "finished", None);
+ let d = LoggingDeleter::new(self.info.clone(), self.logger.clone(), d);
+ (rp, d)
})
.map_err(|err| {
self.logger.log(
&self.info,
Operation::BlockingDelete,
- &[("path", path)],
+ &[],
"failed",
Some(&err),
);
@@ -1351,3 +1290,225 @@ impl<P: oio::BlockingList, I: LoggingInterceptor> oio::BlockingList for LoggingL
res
}
}
+
+pub struct LoggingDeleter<D, I: LoggingInterceptor> {
+ info: Arc<AccessorInfo>,
+ logger: I,
+
+ queued: usize,
+ deleted: usize,
+ inner: D,
+}
+
+impl<D, I: LoggingInterceptor> LoggingDeleter<D, I> {
+ fn new(info: Arc<AccessorInfo>, logger: I, inner: D) -> Self {
+ Self {
+ info,
+ logger,
+
+ queued: 0,
+ deleted: 0,
+ inner,
+ }
+ }
+}
+
+impl<D: oio::Delete, I: LoggingInterceptor> oio::Delete for LoggingDeleter<D, I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let version = args
+ .version()
+ .clone()
+ .map(|v| v.to_string())
+ .unwrap_or_else(|| "<latest>".to_string());
+
+ self.logger.log(
+ &self.info,
+ Operation::DeleterDelete,
+ &[("path", path), ("version", &version)],
+ "started",
+ None,
+ );
+
+ let res = self.inner.delete(path, args);
+
+ match &res {
+ Ok(_) => {
+ self.queued += 1;
+ self.logger.log(
+ &self.info,
+ Operation::DeleterDelete,
+ &[
+ ("path", &path),
+ ("version", &version),
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "succeeded",
+ None,
+ );
+ }
+ Err(err) => {
+ self.logger.log(
+ &self.info,
+ Operation::DeleterDelete,
+ &[
+ ("path", &path),
+ ("version", &version),
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "failed",
+ Some(err),
+ );
+ }
+ };
+
+ res
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ self.logger.log(
+ &self.info,
+ Operation::DeleterFlush,
+ &[
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "started",
+ None,
+ );
+
+ let res = self.inner.flush().await;
+
+ match &res {
+ Ok(flushed) => {
+ self.queued -= flushed;
+ self.deleted += flushed;
+ self.logger.log(
+ &self.info,
+ Operation::DeleterFlush,
+ &[
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "succeeded",
+ None,
+ );
+ }
+ Err(err) => {
+ self.logger.log(
+ &self.info,
+ Operation::DeleterFlush,
+ &[
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "failed",
+ Some(err),
+ );
+ }
+ };
+
+ res
+ }
+}
+
+impl<D: oio::BlockingDelete, I: LoggingInterceptor> oio::BlockingDelete for LoggingDeleter<D, I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let version = args
+ .version()
+ .clone()
+ .map(|v| v.to_string())
+ .unwrap_or_else(|| "<latest>".to_string());
+
+ self.logger.log(
+ &self.info,
+ Operation::BlockingDeleterDelete,
+ &[("path", path), ("version", &version)],
+ "started",
+ None,
+ );
+
+ let res = self.inner.delete(path, args);
+
+ match &res {
+ Ok(_) => {
+ self.queued += 1;
+ self.logger.log(
+ &self.info,
+ Operation::BlockingDeleterDelete,
+ &[
+ ("path", &path),
+ ("version", &version),
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "succeeded",
+ None,
+ );
+ }
+ Err(err) => {
+ self.logger.log(
+ &self.info,
+ Operation::BlockingDeleterDelete,
+ &[
+ ("path", &path),
+ ("version", &version),
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "failed",
+ Some(err),
+ );
+ }
+ };
+
+ res
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ self.logger.log(
+ &self.info,
+ Operation::BlockingDeleterFlush,
+ &[
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "started",
+ None,
+ );
+
+ let res = self.inner.flush();
+
+ match &res {
+ Ok(flushed) => {
+ self.queued -= flushed;
+ self.deleted += flushed;
+ self.logger.log(
+ &self.info,
+ Operation::BlockingDeleterFlush,
+ &[
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "succeeded",
+ None,
+ );
+ }
+ Err(err) => {
+ self.logger.log(
+ &self.info,
+ Operation::BlockingDeleterFlush,
+ &[
+ ("queued", &self.queued.to_string()),
+ ("deleted", &self.deleted.to_string()),
+ ],
+ "failed",
+ Some(err),
+ );
+ }
+ };
+
+ res
+ }
+}
diff --git a/core/src/layers/mime_guess.rs b/core/src/layers/mime_guess.rs
index b9fa1dc8c..62bb6280e 100644
--- a/core/src/layers/mime_guess.rs
+++ b/core/src/layers/mime_guess.rs
@@ -17,6 +17,7 @@
use crate::raw::*;
use crate::Result;
+use std::future::Future;
/// A layer that can automatically set `Content-Type` based on the file extension in the path.
///
@@ -106,27 +107,28 @@ fn rpstat_with_mime(path: &str, rp: RpStat) -> RpStat {
impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
type Inner = A;
type Reader = A::Reader;
- type BlockingReader = A::BlockingReader;
type Writer = A::Writer;
- type BlockingWriter = A::BlockingWriter;
type Lister = A::Lister;
+ type Deleter = A::Deleter;
+ type BlockingReader = A::BlockingReader;
+ type BlockingWriter = A::BlockingWriter;
type BlockingLister = A::BlockingLister;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.0
}
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ self.inner().read(path, args).await
+ }
+
async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
self.inner()
.write(path, opwrite_with_mime(path, args))
.await
}
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- self.inner()
- .blocking_write(path, opwrite_with_mime(path, args))
- }
-
async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
self.inner()
.stat(path, args)
@@ -134,14 +136,8 @@ impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
.map(|rp| rpstat_with_mime(path, rp))
}
- fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
- self.inner()
- .blocking_stat(path, args)
- .map(|rp| rpstat_with_mime(path, rp))
- }
-
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- self.inner().read(path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner().delete().await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -152,6 +148,21 @@ impl<A: Access> LayeredAccess for MimeGuessAccessor<A> {
self.inner().blocking_read(path, args)
}
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
+ self.inner()
+ .blocking_write(path, opwrite_with_mime(path, args))
+ }
+
+ fn blocking_stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ self.inner()
+ .blocking_stat(path, args)
+ .map(|rp| rpstat_with_mime(path, rp))
+ }
+
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner().blocking_delete()
+ }
+
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.inner().blocking_list(path, args)
}
diff --git a/core/src/layers/observe/metrics.rs b/core/src/layers/observe/metrics.rs
index 40d7ddf1b..e1903b824 100644
--- a/core/src/layers/observe/metrics.rs
+++ b/core/src/layers/observe/metrics.rs
@@ -18,6 +18,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
use std::fmt::Write;
+use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
@@ -179,6 +180,8 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
type BlockingWriter = MetricsWrapper<A::BlockingWriter, I>;
type Lister = MetricsWrapper<A::Lister, I>;
type BlockingLister = MetricsWrapper<A::BlockingLister, I>;
+ type Deleter = MetricsWrapper<A::Deleter, I>;
+ type BlockingDeleter = MetricsWrapper<A::BlockingDeleter, I>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -396,35 +399,48 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
})
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
let op = Operation::Delete;
let start = Instant::now();
- self.inner()
- .delete(path, args)
+ let (rp, writer) = self
+ .inner
+ .delete()
.await
.map(|v| {
self.interceptor.observe_operation_duration_seconds(
self.scheme,
self.namespace.clone(),
self.root.clone(),
- path,
+ "",
op,
start.elapsed(),
);
v
})
- .map_err(move |err| {
+ .map_err(|err| {
self.interceptor.observe_operation_errors_total(
self.scheme,
self.namespace.clone(),
self.root.clone(),
- path,
+ "",
op,
err.kind(),
);
err
- })
+ })?;
+
+ Ok((
+ rp,
+ MetricsWrapper::new(
+ writer,
+ self.interceptor.clone(),
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ "".to_string(),
+ ),
+ ))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -471,37 +487,6 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
))
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let op = Operation::Batch;
-
- let start = Instant::now();
- self.inner()
- .batch(args)
- .await
- .map(|v| {
- self.interceptor.observe_operation_duration_seconds(
- self.scheme,
- self.namespace.clone(),
- self.root.clone(),
- "",
- op,
- start.elapsed(),
- );
- v
- })
- .map_err(move |err| {
- self.interceptor.observe_operation_errors_total(
- self.scheme,
- self.namespace.clone(),
- self.root.clone(),
- "",
- op,
- err.kind(),
- );
- err
- })
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
let op = Operation::Presign;
@@ -739,34 +724,47 @@ impl<A: Access, I: MetricsIntercept> LayeredAccess for MetricsAccessor<A, I> {
})
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let op = Operation::BlockingDelete;
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ let op = Operation::Delete;
let start = Instant::now();
- self.inner()
- .blocking_delete(path, args)
+ let (rp, writer) = self
+ .inner
+ .delete()
.map(|v| {
self.interceptor.observe_operation_duration_seconds(
self.scheme,
self.namespace.clone(),
self.root.clone(),
- path,
+ "",
op,
start.elapsed(),
);
v
})
- .map_err(move |err| {
+ .map_err(|err| {
self.interceptor.observe_operation_errors_total(
self.scheme,
self.namespace.clone(),
self.root.clone(),
- path,
+ "",
op,
err.kind(),
);
err
- })
+ })?;
+
+ Ok((
+ rp,
+ MetricsWrapper::new(
+ writer,
+ self.interceptor.clone(),
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ "".to_string(),
+ ),
+ ))
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -1166,3 +1164,127 @@ impl<R: oio::BlockingList, I: MetricsIntercept> oio::BlockingList for MetricsWra
res
}
}
+
+impl<R: oio::Delete, I: MetricsIntercept> oio::Delete for MetricsWrapper<R, I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let op = Operation::DeleterDelete;
+
+ let start = Instant::now();
+
+ let res = match self.inner.delete(path, args) {
+ Ok(entry) => Ok(entry),
+ Err(err) => {
+ self.interceptor.observe_operation_errors_total(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ err.kind(),
+ );
+ Err(err)
+ }
+ };
+ self.interceptor.observe_operation_duration_seconds(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ start.elapsed(),
+ );
+ res
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ let op = Operation::DeleterFlush;
+
+ let start = Instant::now();
+
+ let res = match self.inner.flush().await {
+ Ok(entry) => Ok(entry),
+ Err(err) => {
+ self.interceptor.observe_operation_errors_total(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ err.kind(),
+ );
+ Err(err)
+ }
+ };
+ self.interceptor.observe_operation_duration_seconds(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ start.elapsed(),
+ );
+ res
+ }
+}
+
+impl<R: oio::BlockingDelete, I: MetricsIntercept> oio::BlockingDelete for MetricsWrapper<R, I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ let op = Operation::BlockingDeleterDelete;
+
+ let start = Instant::now();
+
+ let res = match self.inner.delete(&path, args) {
+ Ok(entry) => Ok(entry),
+ Err(err) => {
+ self.interceptor.observe_operation_errors_total(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ err.kind(),
+ );
+ Err(err)
+ }
+ };
+ self.interceptor.observe_operation_duration_seconds(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ start.elapsed(),
+ );
+ res
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ let op = Operation::BlockingDeleterFlush;
+
+ let start = Instant::now();
+
+ let res = match self.inner.flush() {
+ Ok(entry) => Ok(entry),
+ Err(err) => {
+ self.interceptor.observe_operation_errors_total(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ err.kind(),
+ );
+ Err(err)
+ }
+ };
+ self.interceptor.observe_operation_duration_seconds(
+ self.scheme,
+ self.namespace.clone(),
+ self.root.clone(),
+ &self.path,
+ op,
+ start.elapsed(),
+ );
+ res
+ }
+}
diff --git a/core/src/layers/oteltrace.rs b/core/src/layers/oteltrace.rs
index 58cb7e4df..dcea7f93d 100644
--- a/core/src/layers/oteltrace.rs
+++ b/core/src/layers/oteltrace.rs
@@ -72,6 +72,8 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
type BlockingWriter = OtelTraceWrapper<A::BlockingWriter>;
type Lister = OtelTraceWrapper<A::Lister>;
type BlockingLister = OtelTraceWrapper<A::BlockingLister>;
+ type Deleter = A::Deleter;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -142,13 +144,8 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
self.inner().stat(path, args).with_context(cx).await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let tracer = global::tracer("opendal");
- let mut span = tracer.start("delete");
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- let cx = TraceContext::current_with_span(span);
- self.inner().delete(path, args).with_context(cx).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner().delete().await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -162,14 +159,6 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
.map(|(rp, s)| (rp, OtelTraceWrapper::new(span, s)))
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let tracer = global::tracer("opendal");
- let mut span = tracer.start("batch");
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- let cx = TraceContext::current_with_span(span);
- self.inner().batch(args).with_context(cx).await
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
let tracer = global::tracer("opendal");
let mut span = tracer.start("presign");
@@ -241,14 +230,8 @@ impl<A: Access> LayeredAccess for OtelTraceAccessor<A> {
})
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let tracer = global::tracer("opendal");
- tracer.in_span("blocking_delete", |cx| {
- let span = cx.span();
- span.set_attribute(KeyValue::new("path", path.to_string()));
- span.set_attribute(KeyValue::new("args", format!("{:?}", args)));
- self.inner().blocking_delete(path, args)
- })
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner().blocking_delete()
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index af1883ac8..b1d15fa01 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -17,6 +17,7 @@
use std::fmt::Debug;
use std::fmt::Formatter;
+use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
@@ -299,6 +300,8 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
type BlockingWriter = RetryWrapper<A::BlockingWriter, I>;
type Lister = RetryWrapper<A::Lister, I>;
type BlockingLister = RetryWrapper<A::BlockingLister, I>;
+ type Deleter = RetryWrapper<A::Deleter, I>;
+ type BlockingDeleter = RetryWrapper<A::BlockingDeleter, I>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -346,12 +349,13 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
.map_err(|e| e.set_persistent())
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- { || self.inner.delete(path, args.clone()) }
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ { || self.inner.delete() }
.retry(self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| self.notify.intercept(err, dur))
.await
+ .map(|(rp, r)| (rp, RetryWrapper::new(r, self.notify.clone(), self.builder)))
.map_err(|e| e.set_persistent())
}
@@ -379,32 +383,10 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
.when(|e| e.is_temporary())
.notify(|err, dur| self.notify.intercept(err, dur))
.await
- .map(|(l, p)| {
- let lister = RetryWrapper::new(p, self.notify.clone(), self.builder);
- (l, lister)
- })
+ .map(|(rp, r)| (rp, RetryWrapper::new(r, self.notify.clone(), self.builder)))
.map_err(|e| e.set_persistent())
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- {
- || async {
- let rp = self.inner.batch(args.clone()).await?;
- let mut nrp = Vec::with_capacity(rp.results().len());
- for (path, result) in rp.into_results() {
- let result = result?;
- nrp.push((path, Ok(result)))
- }
- Ok(RpBatch::new(nrp))
- }
- }
- .retry(self.builder)
- .when(|e: &Error| e.is_temporary())
- .notify(|err, dur| self.notify.intercept(err, dur))
- .await
- .map_err(|e| e.set_persistent())
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
{ || self.inner.blocking_create_dir(path, args.clone()) }
.retry(self.builder)
@@ -447,12 +429,13 @@ impl<A: Access, I: RetryInterceptor> LayeredAccess for RetryAccessor<A, I> {
.map_err(|e| e.set_persistent())
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- { || self.inner.blocking_delete(path, args.clone()) }
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ { || self.inner.blocking_delete() }
.retry(self.builder)
.when(|e| e.is_temporary())
.notify(|err, dur| self.notify.intercept(err, dur))
.call()
+ .map(|(rp, r)| (rp, RetryWrapper::new(r, self.notify.clone(), self.builder)))
.map_err(|e| e.set_persistent())
}
@@ -751,6 +734,65 @@ impl<P: oio::BlockingList, I: RetryInterceptor> oio::BlockingList for RetryWrapp
}
}
+impl<P: oio::Delete, I: RetryInterceptor> oio::Delete for RetryWrapper<P, I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ { || self.inner.as_mut().unwrap().delete(path, args.clone()) }
+ .retry(self.builder)
+ .when(|e| e.is_temporary())
+ .notify(|err, dur| {
+ self.notify.intercept(err, dur);
+ })
+ .call()
+ .map_err(|e| e.set_persistent())
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ use backon::RetryableWithContext;
+
+ let inner = self.take_inner()?;
+
+ let (inner, res) = {
+ |mut p: P| async move {
+ let res = p.flush().await;
+
+ (p, res)
+ }
+ }
+ .retry(self.builder)
+ .when(|e| e.is_temporary())
+ .context(inner)
+ .notify(|err, dur| self.notify.intercept(err, dur))
+ .await;
+
+ self.inner = Some(inner);
+ res.map_err(|err| err.set_persistent())
+ }
+}
+
+impl<P: oio::BlockingDelete, I: RetryInterceptor> oio::BlockingDelete for RetryWrapper<P, I> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ { || self.inner.as_mut().unwrap().delete(path, args.clone()) }
+ .retry(self.builder)
+ .when(|e| e.is_temporary())
+ .notify(|err, dur| {
+ self.notify.intercept(err, dur);
+ })
+ .call()
+ .map_err(|e| e.set_persistent())
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ { || self.inner.as_mut().unwrap().flush() }
+ .retry(self.builder)
+ .when(|e| e.is_temporary())
+ .notify(|err, dur| {
+ self.notify.intercept(err, dur);
+ })
+ .call()
+ .map_err(|e| e.set_persistent())
+ }
+}
+
#[cfg(test)]
mod tests {
use std::sync::Arc;
@@ -788,9 +830,11 @@ mod tests {
type Reader = MockReader;
type Writer = MockWriter;
type Lister = MockLister;
+ type Deleter = ();
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -833,63 +877,6 @@ mod tests {
let lister = MockLister::default();
Ok((RpList::default(), lister))
}
-
- async fn batch(&self, op: OpBatch) -> Result<RpBatch> {
- let mut attempt = self.attempt.lock().unwrap();
- *attempt += 1;
-
- match *attempt {
- 1 => Err(
- Error::new(ErrorKind::Unexpected, "retryable_error from reader")
- .set_temporary(),
- ),
- 2 => Ok(RpBatch::new(
- op.into_operation()
- .into_iter()
- .map(|(s, _)| {
- (
- s,
- Err(Error::new(
- ErrorKind::Unexpected,
- "retryable_error from reader",
- )
- .set_temporary()),
- )
- })
- .collect(),
- )),
- 3 => Ok(RpBatch::new(
- op.into_operation()
- .into_iter()
- .enumerate()
- .map(|(i, (s, _))| {
- (
- s,
- match i {
- 0 => Err(Error::new(
- ErrorKind::Unexpected,
- "retryable_error from reader",
- )
- .set_temporary()),
- _ => Ok(RpDelete {}.into()),
- },
- )
- })
- .collect(),
- )),
- 4 => Err(
- Error::new(ErrorKind::Unexpected, "retryable_error from reader")
- .set_temporary(),
- ),
- 5 => Ok(RpBatch::new(
- op.into_operation()
- .into_iter()
- .map(|(s, _)| (s, Ok(RpDelete {}.into())))
- .collect(),
- )),
- _ => unreachable!(),
- }
- }
}
#[derive(Debug, Clone, Default)]
diff --git a/core/src/layers/throttle.rs b/core/src/layers/throttle.rs
index 06a2f6313..8ffeb0edb 100644
--- a/core/src/layers/throttle.rs
+++ b/core/src/layers/throttle.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::num::NonZeroU32;
use std::sync::Arc;
use std::thread;
@@ -114,11 +115,13 @@ pub struct ThrottleAccessor<A: Access> {
impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
type Inner = A;
type Reader = ThrottleWrapper<A::Reader>;
- type BlockingReader = ThrottleWrapper<A::BlockingReader>;
type Writer = ThrottleWrapper<A::Writer>;
- type BlockingWriter = ThrottleWrapper<A::BlockingWriter>;
type Lister = A::Lister;
+ type Deleter = A::Deleter;
+ type BlockingReader = ThrottleWrapper<A::BlockingReader>;
+ type BlockingWriter = ThrottleWrapper<A::BlockingWriter>;
type BlockingLister = A::BlockingLister;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -142,6 +145,10 @@ impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
.map(|(rp, w)| (rp, ThrottleWrapper::new(w, limiter)))
}
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner.delete().await
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner.list(path, args).await
}
@@ -162,6 +169,10 @@ impl<A: Access> LayeredAccess for ThrottleAccessor<A> {
.map(|(rp, w)| (rp, ThrottleWrapper::new(w, limiter)))
}
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner.blocking_delete()
+ }
+
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.inner.blocking_list(path, args)
}
diff --git a/core/src/layers/timeout.rs b/core/src/layers/timeout.rs
index 2bda1292b..e3c96e96c 100644
--- a/core/src/layers/timeout.rs
+++ b/core/src/layers/timeout.rs
@@ -220,6 +220,8 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
type BlockingWriter = A::BlockingWriter;
type Lister = TimeoutWrapper<A::Lister>;
type BlockingLister = A::BlockingLister;
+ type Deleter = TimeoutWrapper<A::Deleter>;
+ type BlockingDeleter = A::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -271,9 +273,10 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
.await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.timeout(Operation::Delete, self.inner.delete(path, args))
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.io_timeout(Operation::Write, self.inner.delete())
.await
+ .map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -282,10 +285,6 @@ impl<A: Access> LayeredAccess for TimeoutAccessor<A> {
.map(|(rp, r)| (rp, TimeoutWrapper::new(r, self.io_timeout)))
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.timeout(Operation::Batch, self.inner.batch(args)).await
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
self.timeout(Operation::Presign, self.inner.presign(path, args))
.await
@@ -382,6 +381,17 @@ impl<R: oio::List> oio::List for TimeoutWrapper<R> {
}
}
+impl<R: oio::Delete> oio::Delete for TimeoutWrapper<R> {
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ let fut = self.inner.flush();
+ Self::io_timeout(self.timeout, Operation::DeleterFlush.into_static(), fut).await
+ }
+}
+
#[cfg(test)]
mod tests {
use std::future::pending;
@@ -408,6 +418,8 @@ mod tests {
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type Deleter = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -426,10 +438,10 @@ mod tests {
}
/// This function will never return.
- async fn delete(&self, _: &str, _: OpDelete) -> Result<RpDelete> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
sleep(Duration::from_secs(u64::MAX)).await;
- Ok(RpDelete::default())
+ Ok((RpDelete::default(), ()))
}
async fn list(&self, _: &str, _: OpList) -> Result<(RpList, Self::Lister)> {
diff --git a/core/src/layers/tracing.rs b/core/src/layers/tracing.rs
index c59f8d5fc..2a33c0c2a 100644
--- a/core/src/layers/tracing.rs
+++ b/core/src/layers/tracing.rs
@@ -154,11 +154,13 @@ pub struct TracingAccessor<A> {
impl<A: Access> LayeredAccess for TracingAccessor<A> {
type Inner = A;
type Reader = TracingWrapper<A::Reader>;
- type BlockingReader = TracingWrapper<A::BlockingReader>;
type Writer = TracingWrapper<A::Writer>;
- type BlockingWriter = TracingWrapper<A::BlockingWriter>;
type Lister = TracingWrapper<A::Lister>;
+ type Deleter = TracingWrapper<A::Deleter>;
+ type BlockingReader = TracingWrapper<A::BlockingReader>;
+ type BlockingWriter = TracingWrapper<A::BlockingWriter>;
type BlockingLister = TracingWrapper<A::BlockingLister>;
+ type BlockingDeleter = TracingWrapper<A::BlockingDeleter>;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -206,8 +208,11 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
}
#[tracing::instrument(level = "debug", skip(self))]
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.delete(path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner
+ .delete()
+ .await
+ .map(|(rp, r)| (rp, TracingWrapper::new(Span::current(), r)))
}
#[tracing::instrument(level = "debug", skip(self))]
@@ -223,11 +228,6 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
self.inner.presign(path, args).await
}
- #[tracing::instrument(level = "debug", skip(self))]
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- self.inner.batch(args).await
- }
-
#[tracing::instrument(level = "debug", skip(self))]
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.inner.blocking_create_dir(path, args)
@@ -263,8 +263,10 @@ impl<A: Access> LayeredAccess for TracingAccessor<A> {
}
#[tracing::instrument(level = "debug", skip(self))]
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner.blocking_delete(path, args)
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner
+ .blocking_delete()
+ .map(|(rp, r)| (rp, TracingWrapper::new(Span::current(), r)))
}
#[tracing::instrument(level = "debug", skip(self))]
@@ -363,3 +365,27 @@ impl<R: oio::BlockingList> oio::BlockingList for TracingWrapper<R> {
self.inner.next()
}
}
+
+impl<R: oio::Delete> oio::Delete for TracingWrapper<R> {
+ #[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ #[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
+ async fn flush(&mut self) -> Result<usize> {
+ self.inner.flush().await
+ }
+}
+
+impl<R: oio::BlockingDelete> oio::BlockingDelete for TracingWrapper<R> {
+ #[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ self.inner.delete(path, args)
+ }
+
+ #[tracing::instrument(parent = &self.span, level = "debug", skip_all)]
+ fn flush(&mut self) -> Result<usize> {
+ self.inner.flush()
+ }
+}
diff --git a/core/src/layers/type_eraser.rs b/core/src/layers/type_eraser.rs
index 09e0a7fb1..b9a0567b9 100644
--- a/core/src/layers/type_eraser.rs
+++ b/core/src/layers/type_eraser.rs
@@ -15,11 +15,11 @@
// specific language governing permissions and limitations
// under the License.
-use std::fmt::Debug;
-use std::fmt::Formatter;
-
use crate::raw::*;
use crate::*;
+use std::fmt::Debug;
+use std::fmt::Formatter;
+use std::future::Future;
/// TypeEraseLayer will erase the types on internal accessor.
///
@@ -53,11 +53,13 @@ impl<A: Access> Debug for TypeEraseAccessor<A> {
impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
type Inner = A;
type Reader = oio::Reader;
- type BlockingReader = oio::BlockingReader;
type Writer = oio::Writer;
- type BlockingWriter = oio::BlockingWriter;
type Lister = oio::Lister;
+ type Deleter = oio::Deleter;
+ type BlockingReader = oio::BlockingReader;
+ type BlockingWriter = oio::BlockingWriter;
type BlockingLister = oio::BlockingLister;
+ type BlockingDeleter = oio::BlockingDeleter;
fn inner(&self) -> &Self::Inner {
&self.inner
@@ -77,6 +79,13 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
.map(|(rp, w)| (rp, Box::new(w) as oio::Writer))
}
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.inner
+ .delete()
+ .await
+ .map(|(rp, p)| (rp, Box::new(p) as oio::Deleter))
+ }
+
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
self.inner
.list(path, args)
@@ -96,6 +105,12 @@ impl<A: Access> LayeredAccess for TypeEraseAccessor<A> {
.map(|(rp, w)| (rp, Box::new(w) as oio::BlockingWriter))
}
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner
+ .blocking_delete()
+ .map(|(rp, p)| (rp, Box::new(p) as oio::BlockingDeleter))
+ }
+
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
self.inner
.blocking_list(path, args)
diff --git a/core/src/raw/accessor.rs b/core/src/raw/accessor.rs
index 888dc4fa1..a7e4aa08b 100644
--- a/core/src/raw/accessor.rs
+++ b/core/src/raw/accessor.rs
@@ -61,6 +61,8 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
type Writer: oio::Write;
/// Lister is the associated lister returned in `list` operation.
type Lister: oio::List;
+ /// Deleter is the associated deleter returned in `delete` operation.
+ type Deleter: oio::Delete;
/// BlockingReader is the associated reader returned `blocking_read` operation.
type BlockingReader: oio::BlockingRead;
@@ -68,6 +70,8 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
type BlockingWriter: oio::BlockingWrite;
/// BlockingLister is the associated lister returned `blocking_list` operation.
type BlockingLister: oio::BlockingList;
+ /// BlockingDeleter is the associated deleter returned `blocking_delete` operation.
+ type BlockingDeleter: oio::BlockingDelete;
/// Invoke the `info` operation to get metadata of accessor.
///
@@ -172,13 +176,7 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
///
/// - `delete` is an idempotent operation, it's safe to call `Delete` on the same path multiple times.
/// - `delete` SHOULD return `Ok(())` if the path is deleted successfully or not exist.
- fn delete(
- &self,
- path: &str,
- args: OpDelete,
- ) -> impl Future<Output = Result<RpDelete>> + MaybeSend {
- let (_, _) = (path, args);
-
+ fn delete(&self) -> impl Future<Output = Result<(RpDelete, Self::Deleter)>> + MaybeSend {
ready(Err(Error::new(
ErrorKind::Unsupported,
"operation is not supported",
@@ -266,18 +264,6 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
)))
}
- /// Invoke the `batch` operations.
- ///
- /// Require [`Capability::batch`]
- fn batch(&self, args: OpBatch) -> impl Future<Output = Result<RpBatch>> + MaybeSend {
- let _ = args;
-
- ready(Err(Error::new(
- ErrorKind::Unsupported,
- "operation is not supported",
- )))
- }
-
/// Invoke the `blocking_create` operation on the specified path.
///
/// This operation is the blocking version of [`Accessor::create_dir`]
@@ -339,9 +325,7 @@ pub trait Access: Send + Sync + Debug + Unpin + 'static {
/// This operation is the blocking version of [`Accessor::delete`]
///
/// Require [`Capability::write`] and [`Capability::blocking`]
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- let (_, _) = (path, args);
-
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
Err(Error::new(
ErrorKind::Unsupported,
"operation is not supported",
@@ -421,8 +405,7 @@ pub trait AccessDyn: Send + Sync + Debug + Unpin {
args: OpWrite,
) -> BoxedFuture<'a, Result<(RpWrite, oio::Writer)>>;
/// Dyn version of [`Accessor::delete`]
- fn delete_dyn<'a>(&'a self, path: &'a str, args: OpDelete)
- -> BoxedFuture<'a, Result<RpDelete>>;
+ fn delete_dyn(&self) -> BoxedFuture<Result<(RpDelete, oio::Deleter)>>;
/// Dyn version of [`Accessor::list`]
fn list_dyn<'a>(
&'a self,
@@ -449,8 +432,6 @@ pub trait AccessDyn: Send + Sync + Debug + Unpin {
path: &'a str,
args: OpPresign,
) -> BoxedFuture<'a, Result<RpPresign>>;
- /// Dyn version of [`Accessor::batch`]
- fn batch_dyn(&self, args: OpBatch) -> BoxedFuture<'_, Result<RpBatch>>;
/// Dyn version of [`Accessor::blocking_create_dir`]
fn blocking_create_dir_dyn(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir>;
/// Dyn version of [`Accessor::blocking_stat`]
@@ -464,7 +445,7 @@ pub trait AccessDyn: Send + Sync + Debug + Unpin {
args: OpWrite,
) -> Result<(RpWrite, oio::BlockingWriter)>;
/// Dyn version of [`Accessor::blocking_delete`]
- fn blocking_delete_dyn(&self, path: &str, args: OpDelete) -> Result<RpDelete>;
+ fn blocking_delete_dyn(&self) -> Result<(RpDelete, oio::BlockingDeleter)>;
/// Dyn version of [`Accessor::blocking_list`]
fn blocking_list_dyn(&self, path: &str, args: OpList) -> Result<(RpList, oio::BlockingLister)>;
/// Dyn version of [`Accessor::blocking_copy`]
@@ -516,12 +497,8 @@ where
Box::pin(self.write(path, args))
}
- fn delete_dyn<'a>(
- &'a self,
- path: &'a str,
- args: OpDelete,
- ) -> BoxedFuture<'a, Result<RpDelete>> {
- Box::pin(self.delete(path, args))
+ fn delete_dyn(&self) -> BoxedFuture<Result<(RpDelete, oio::Deleter)>> {
+ Box::pin(self.delete())
}
fn list_dyn<'a>(
@@ -558,10 +535,6 @@ where
Box::pin(self.presign(path, args))
}
- fn batch_dyn(&self, args: OpBatch) -> BoxedFuture<'_, Result<RpBatch>> {
- Box::pin(self.batch(args))
- }
-
fn blocking_create_dir_dyn(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.blocking_create_dir(path, args)
}
@@ -582,8 +555,8 @@ where
self.blocking_write(path, args)
}
- fn blocking_delete_dyn(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.blocking_delete(path, args)
+ fn blocking_delete_dyn(&self) -> Result<(RpDelete, oio::Deleter)> {
+ self.blocking_delete()
}
fn blocking_list_dyn(&self, path: &str, args: OpList) -> Result<(RpList, oio::BlockingLister)> {
@@ -603,9 +576,11 @@ impl Access for dyn AccessDyn {
type Reader = oio::Reader;
type BlockingReader = oio::BlockingReader;
type Writer = oio::Writer;
+ type Deleter = oio::Deleter;
type BlockingWriter = oio::BlockingWriter;
type Lister = oio::Lister;
type BlockingLister = oio::BlockingLister;
+ type BlockingDeleter = oio::BlockingDeleter;
fn info(&self) -> Arc<AccessorInfo> {
self.info_dyn()
@@ -627,8 +602,8 @@ impl Access for dyn AccessDyn {
self.write_dyn(path, args).await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.delete_dyn(path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ self.delete_dyn().await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -647,10 +622,6 @@ impl Access for dyn AccessDyn {
self.presign_dyn(path, args).await
}
- fn batch(&self, args: OpBatch) -> impl Future<Output = Result<RpBatch>> + MaybeSend {
- self.batch_dyn(args)
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.blocking_create_dir_dyn(path, args)
}
@@ -667,8 +638,8 @@ impl Access for dyn AccessDyn {
self.blocking_write_dyn(path, args)
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.blocking_delete_dyn(path, args)
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.blocking_delete_dyn()
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -689,9 +660,11 @@ impl Access for () {
type Reader = ();
type Writer = ();
type Lister = ();
+ type Deleter = ();
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
AccessorInfo {
@@ -714,9 +687,11 @@ impl<T: Access + ?Sized> Access for Arc<T> {
type Reader = T::Reader;
type Writer = T::Writer;
type Lister = T::Lister;
+ type Deleter = T::Deleter;
type BlockingReader = T::BlockingReader;
type BlockingWriter = T::BlockingWriter;
type BlockingLister = T::BlockingLister;
+ type BlockingDeleter = T::BlockingDeleter;
fn info(&self) -> Arc<AccessorInfo> {
self.as_ref().info()
@@ -750,12 +725,8 @@ impl<T: Access + ?Sized> Access for Arc<T> {
async move { self.as_ref().write(path, args).await }
}
- fn delete(
- &self,
- path: &str,
- args: OpDelete,
- ) -> impl Future<Output = Result<RpDelete>> + MaybeSend {
- async move { self.as_ref().delete(path, args).await }
+ fn delete(&self) -> impl Future<Output = Result<(RpDelete, Self::Deleter)>> + MaybeSend {
+ async move { self.as_ref().delete().await }
}
fn list(
@@ -792,10 +763,6 @@ impl<T: Access + ?Sized> Access for Arc<T> {
async move { self.as_ref().presign(path, args).await }
}
- fn batch(&self, args: OpBatch) -> impl Future<Output = Result<RpBatch>> + MaybeSend {
- async move { self.as_ref().batch(args).await }
- }
-
fn blocking_create_dir(&self, path: &str, args: OpCreateDir) -> Result<RpCreateDir> {
self.as_ref().blocking_create_dir(path, args)
}
@@ -812,8 +779,8 @@ impl<T: Access + ?Sized> Access for Arc<T> {
self.as_ref().blocking_write(path, args)
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.as_ref().blocking_delete(path, args)
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.as_ref().blocking_delete()
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
diff --git a/core/src/raw/adapters/kv/backend.rs b/core/src/raw/adapters/kv/backend.rs
index a2f378a46..23f0e951b 100644
--- a/core/src/raw/adapters/kv/backend.rs
+++ b/core/src/raw/adapters/kv/backend.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use std::future::Future;
use std::sync::Arc;
use std::vec::IntoIter;
@@ -65,11 +66,13 @@ where
impl<S: Adapter> Access for Backend<S> {
type Reader = Buffer;
- type BlockingReader = Buffer;
type Writer = KvWriter<S>;
- type BlockingWriter = KvWriter<S>;
type Lister = HierarchyLister<KvLister<S::Scanner>>;
+ type Deleter = oio::OneShotDeleter<S>;
+ type BlockingReader = Buffer;
+ type BlockingWriter = KvWriter<S>;
type BlockingLister = HierarchyLister<BlockingKvLister>;
+ type BlockingDeleter = oio::OneShotDeleter<S>;
fn info(&self) -> Arc<AccessorInfo> {
let kv_info = self.kv.info();
@@ -97,18 +100,25 @@ impl<S: Adapter> Access for Backend<S> {
am.into()
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
let p = build_abs_path(&self.root, path);
- let bs = match self.kv.get(&p).await? {
- Some(bs) => bs,
- None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
- };
- Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
+
+ if p == build_abs_path(&self.root, "") {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ let bs = self.kv.get(&p).await?;
+ match bs {
+ Some(bs) => Ok(RpStat::new(
+ Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
+ )),
+ None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ }
+ }
}
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let p = build_abs_path(&self.root, path);
- let bs = match self.kv.blocking_get(&p)? {
+ let bs = match self.kv.get(&p).await? {
Some(bs) => bs,
None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
};
@@ -121,26 +131,20 @@ impl<S: Adapter> Access for Backend<S> {
Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p)))
}
- fn blocking_write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let p = build_abs_path(&self.root, path);
-
- Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p)))
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(KvDeleter::new(self.kv.clone(), self.root.clone())),
+ ))
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
let p = build_abs_path(&self.root, path);
+ let res = self.kv.scan(&p).await?;
+ let lister = KvLister::new(&self.root, res);
+ let lister = HierarchyLister::new(lister, path, args.recursive());
- if p == build_abs_path(&self.root, "") {
- Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
- } else {
- let bs = self.kv.get(&p).await?;
- match bs {
- Some(bs) => Ok(RpStat::new(
- Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64),
- )),
- None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
- }
- }
+ Ok((RpList::default(), lister))
}
fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
@@ -159,27 +163,26 @@ impl<S: Adapter> Access for Backend<S> {
}
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
let p = build_abs_path(&self.root, path);
-
- self.kv.delete(&p).await?;
- Ok(RpDelete::default())
+ let bs = match self.kv.blocking_get(&p)? {
+ Some(bs) => bs,
+ None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ };
+ Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
}
- fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ fn blocking_write(&self, path: &str, _: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
let p = build_abs_path(&self.root, path);
- self.kv.blocking_delete(&p)?;
- Ok(RpDelete::default())
+ Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p)))
}
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let p = build_abs_path(&self.root, path);
- let res = self.kv.scan(&p).await?;
- let lister = KvLister::new(&self.root, res);
- let lister = HierarchyLister::new(lister, path, args.recursive());
-
- Ok((RpList::default(), lister))
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(KvDeleter::new(self.kv.clone(), self.root.clone())),
+ ))
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -318,3 +321,32 @@ impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
Ok(())
}
}
+
+pub struct KvDeleter<S> {
+ kv: Arc<S>,
+ root: String,
+}
+
+impl<S> KvDeleter<S> {
+ fn new(kv: Arc<S>, root: String) -> Self {
+ KvDeleter { kv, root }
+ }
+}
+
+impl<S: Adapter> oio::OneShotDelete for KvDeleter<S> {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+
+ self.kv.delete(&p).await?;
+ Ok(())
+ }
+}
+
+impl<S: Adapter> oio::BlockingOneShotDelete for KvDeleter<S> {
+ fn blocking_delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+
+ self.kv.blocking_delete(&p)?;
+ Ok(())
+ }
+}
diff --git a/core/src/raw/adapters/typed_kv/backend.rs b/core/src/raw/adapters/typed_kv/backend.rs
index 59d10a47e..a96836104 100644
--- a/core/src/raw/adapters/typed_kv/backend.rs
+++ b/core/src/raw/adapters/typed_kv/backend.rs
@@ -53,11 +53,13 @@ where
impl<S: Adapter> Access for Backend<S> {
type Reader = Buffer;
- type BlockingReader = Buffer;
type Writer = KvWriter<S>;
- type BlockingWriter = KvWriter<S>;
type Lister = HierarchyLister<KvLister>;
+ type Deleter = oio::OneShotDeleter<S>;
+ type BlockingReader = Buffer;
+ type BlockingWriter = KvWriter<S>;
type BlockingLister = HierarchyLister<KvLister>;
+ type BlockingDeleter = oio::OneShotDeleter<S>;
fn info(&self) -> Arc<AccessorInfo> {
let kv_info = self.kv.info();
@@ -94,22 +96,24 @@ impl<S: Adapter> Access for Backend<S> {
am.into()
}
- async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
+ async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
let p = build_abs_path(&self.root, path);
- let bs = match self.kv.get(&p).await? {
- // TODO: we can reuse the metadata in value to build content range.
- Some(bs) => bs.value,
- None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
- };
-
- Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
+ if p == build_abs_path(&self.root, "") {
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
+ } else {
+ let bs = self.kv.get(&p).await?;
+ match bs {
+ Some(bs) => Ok(RpStat::new(bs.metadata)),
+ None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ }
+ }
}
- fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
+ async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
let p = build_abs_path(&self.root, path);
- let bs = match self.kv.blocking_get(&p)? {
+ let bs = match self.kv.get(&p).await? {
// TODO: we can reuse the metadata in value to build content range.
Some(bs) => bs.value,
None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
@@ -124,24 +128,20 @@ impl<S: Adapter> Access for Backend<S> {
Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
}
- fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
- let p = build_abs_path(&self.root, path);
-
- Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(KvDeleter::new(self.kv.clone(), self.root.clone())),
+ ))
}
- async fn stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
+ async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
let p = build_abs_path(&self.root, path);
+ let res = self.kv.scan(&p).await?;
+ let lister = KvLister::new(&self.root, res);
+ let lister = HierarchyLister::new(lister, path, args.recursive());
- if p == build_abs_path(&self.root, "") {
- Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
- } else {
- let bs = self.kv.get(&p).await?;
- match bs {
- Some(bs) => Ok(RpStat::new(bs.metadata)),
- None => Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
- }
- }
+ Ok((RpList::default(), lister))
}
fn blocking_stat(&self, path: &str, _: OpStat) -> Result<RpStat> {
@@ -158,27 +158,29 @@ impl<S: Adapter> Access for Backend<S> {
}
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
let p = build_abs_path(&self.root, path);
- self.kv.delete(&p).await?;
- Ok(RpDelete::default())
+ let bs = match self.kv.blocking_get(&p)? {
+ // TODO: we can reuse the metadata in value to build content range.
+ Some(bs) => bs.value,
+ None => return Err(Error::new(ErrorKind::NotFound, "kv doesn't have this path")),
+ };
+
+ Ok((RpRead::new(), bs.slice(args.range().to_range_as_usize())))
}
- fn blocking_delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+ fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
let p = build_abs_path(&self.root, path);
- self.kv.blocking_delete(&p)?;
- Ok(RpDelete::default())
+ Ok((RpWrite::new(), KvWriter::new(self.kv.clone(), p, args)))
}
- async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let p = build_abs_path(&self.root, path);
- let res = self.kv.scan(&p).await?;
- let lister = KvLister::new(&self.root, res);
- let lister = HierarchyLister::new(lister, path, args.recursive());
-
- Ok((RpList::default(), lister))
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(KvDeleter::new(self.kv.clone(), self.root.clone())),
+ ))
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -327,3 +329,32 @@ impl<S: Adapter> oio::BlockingWrite for KvWriter<S> {
Ok(())
}
}
+
+pub struct KvDeleter<S> {
+ kv: Arc<S>,
+ root: String,
+}
+
+impl<S> KvDeleter<S> {
+ fn new(kv: Arc<S>, root: String) -> Self {
+ KvDeleter { kv, root }
+ }
+}
+
+impl<S: crate::raw::adapters::kv::Adapter> oio::OneShotDelete for KvDeleter<S> {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+
+ self.kv.delete(&p).await?;
+ Ok(())
+ }
+}
+
+impl<S: crate::raw::adapters::kv::Adapter> oio::BlockingOneShotDelete for KvDeleter<S> {
+ fn blocking_delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let p = build_abs_path(&self.root, &path);
+
+ self.kv.blocking_delete(&p)?;
+ Ok(())
+ }
+}
diff --git a/core/src/raw/http_util/header.rs b/core/src/raw/http_util/header.rs
index 49dff81a0..6c8ba65db 100644
--- a/core/src/raw/http_util/header.rs
+++ b/core/src/raw/http_util/header.rs
@@ -110,6 +110,11 @@ pub fn parse_content_disposition(headers: &HeaderMap) -> Result<Option<&str>> {
parse_header_to_str(headers, CONTENT_DISPOSITION)
}
+/// Parse multipart boundary from header map.
+pub fn parse_multipart_boundary(headers: &HeaderMap) -> Result<Option<&str>> {
+ parse_header_to_str(headers, CONTENT_TYPE).map(|v| v.and_then(|v| v.split("boundary=").nth(1)))
+}
+
/// Parse header value to string according to name.
#[inline]
pub fn parse_header_to_str<K>(headers: &HeaderMap, name: K) -> Result<Option<&str>>
@@ -324,4 +329,24 @@ mod tests {
assert_eq!(actual, expected)
}
}
+
+ #[test]
+ fn test_parse_multipart_boundary() {
+ let cases = vec![
+ (
+ "multipart/mixed; boundary=gc0p4Jq0M2Yt08jU534c0p",
+ Some("gc0p4Jq0M2Yt08jU534c0p"),
+ ),
+ ("multipart/mixed", None),
+ ];
+
+ for (input, expected) in cases {
+ let mut headers = HeaderMap::new();
+ headers.insert(CONTENT_TYPE, HeaderValue::from_str(input).unwrap());
+
+ let actual = parse_multipart_boundary(&headers).expect("parse must success");
+
+ assert_eq!(actual, expected)
+ }
+ }
}
diff --git a/core/src/raw/http_util/mod.rs b/core/src/raw/http_util/mod.rs
index 8999b6b80..c90b1e485 100644
--- a/core/src/raw/http_util/mod.rs
+++ b/core/src/raw/http_util/mod.rs
@@ -49,6 +49,7 @@ pub use header::parse_header_to_str;
pub use header::parse_into_metadata;
pub use header::parse_last_modified;
pub use header::parse_location;
+pub use header::parse_multipart_boundary;
pub use header::parse_prefixed_headers;
mod uri;
diff --git a/core/src/raw/layer.rs b/core/src/raw/layer.rs
index 6e6d2b802..b346a3df4 100644
--- a/core/src/raw/layer.rs
+++ b/core/src/raw/layer.rs
@@ -62,6 +62,8 @@ use crate::*;
/// type BlockingWriter = A::BlockingWriter;
/// type Lister = A::Lister;
/// type BlockingLister = A::BlockingLister;
+/// type Deleter = A::Deleter;
+/// type BlockingDeleter = A::BlockingDeleter;
///
/// fn inner(&self) -> &Self::Inner {
/// &self.inner
@@ -102,6 +104,14 @@ use crate::*;
/// ) -> Result<(RpList, Self::BlockingLister)> {
/// self.inner.blocking_list(path, args)
/// }
+///
+/// async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+/// self.inner.delete().await
+/// }
+///
+/// fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+/// self.inner.blocking_delete()
+/// }
/// }
///
/// /// The public struct that exposed to users.
@@ -130,12 +140,15 @@ pub trait Layer<A: Access> {
#[allow(missing_docs)]
pub trait LayeredAccess: Send + Sync + Debug + Unpin + 'static {
type Inner: Access;
+
type Reader: oio::Read;
- type BlockingReader: oio::BlockingRead;
type Writer: oio::Write;
- type BlockingWriter: oio::BlockingWrite;
type Lister: oio::List;
+ type Deleter: oio::Delete;
+ type BlockingReader: oio::BlockingRead;
+ type BlockingWriter: oio::BlockingWrite;
type BlockingLister: oio::BlockingList;
+ type BlockingDeleter: oio::BlockingDelete;
fn inner(&self) -> &Self::Inner;
@@ -185,12 +198,8 @@ pub trait LayeredAccess: Send + Sync + Debug + Unpin + 'static {
self.inner().stat(path, args)
}
- fn delete(
- &self,
- path: &str,
- args: OpDelete,
- ) -> impl Future<Output = Result<RpDelete>> + MaybeSend {
- self.inner().delete(path, args)
+ fn delete(&self) -> impl Future<Output = Result<(RpDelete, Self::Deleter)>> + MaybeSend {
+ self.inner().delete()
}
fn list(
@@ -199,10 +208,6 @@ pub trait LayeredAccess: Send + Sync + Debug + Unpin + 'static {
args: OpList,
) -> impl Future<Output = Result<(RpList, Self::Lister)>> + MaybeSend;
- fn batch(&self, args: OpBatch) -> impl Future<Output = Result<RpBatch>> + MaybeSend {
- self.inner().batch(args)
- }
-
fn presign(
&self,
path: &str,
@@ -231,8 +236,8 @@ pub trait LayeredAccess: Send + Sync + Debug + Unpin + 'static {
self.inner().blocking_stat(path, args)
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- self.inner().blocking_delete(path, args)
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
+ self.inner().blocking_delete()
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)>;
@@ -242,9 +247,12 @@ impl<L: LayeredAccess> Access for L {
type Reader = L::Reader;
type Writer = L::Writer;
type Lister = L::Lister;
+ type Deleter = L::Deleter;
+
type BlockingReader = L::BlockingReader;
type BlockingWriter = L::BlockingWriter;
type BlockingLister = L::BlockingLister;
+ type BlockingDeleter = L::BlockingDeleter;
fn info(&self) -> Arc<AccessorInfo> {
LayeredAccess::info(self)
@@ -274,18 +282,14 @@ impl<L: LayeredAccess> Access for L {
LayeredAccess::stat(self, path, args).await
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- LayeredAccess::delete(self, path, args).await
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ LayeredAccess::delete(self).await
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
LayeredAccess::list(self, path, args).await
}
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- LayeredAccess::batch(self, args).await
- }
-
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
LayeredAccess::presign(self, path, args).await
}
@@ -314,8 +318,8 @@ impl<L: LayeredAccess> Access for L {
LayeredAccess::blocking_stat(self, path, args)
}
- fn blocking_delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- LayeredAccess::blocking_delete(self, path, args)
+ fn blocking_delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ LayeredAccess::blocking_delete(self)
}
fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
@@ -336,7 +340,7 @@ mod tests {
struct Test<A: Access> {
#[allow(dead_code)]
inner: Option<A>,
- deleted: Arc<Mutex<bool>>,
+ stated: Arc<Mutex<bool>>,
}
impl<A: Access> Layer<A> for &Test<A> {
@@ -345,7 +349,7 @@ mod tests {
fn layer(&self, inner: A) -> Self::LayeredAccess {
Test {
inner: Some(inner),
- deleted: self.deleted.clone(),
+ stated: self.stated.clone(),
}
}
}
@@ -357,6 +361,8 @@ mod tests {
type BlockingWriter = ();
type Lister = ();
type BlockingLister = ();
+ type Deleter = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -364,14 +370,14 @@ mod tests {
am.into()
}
- async fn delete(&self, _: &str, _: OpDelete) -> Result<RpDelete> {
- let mut x = self.deleted.lock().await;
+ async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
+ let mut x = self.stated.lock().await;
*x = true;
assert!(self.inner.is_some());
// We will not call anything here to test the layer.
- Ok(RpDelete::default())
+ Ok(RpStat::new(Metadata::new(EntryMode::DIR)))
}
}
@@ -379,7 +385,7 @@ mod tests {
async fn test_layer() {
let test = Test {
inner: None,
- deleted: Arc::new(Mutex::new(false)),
+ stated: Arc::new(Mutex::new(false)),
};
let op = Operator::new(Memory::default())
@@ -387,8 +393,8 @@ mod tests {
.layer(&test)
.finish();
- op.delete("xxxxx").await.unwrap();
+ op.stat("xxxxx").await.unwrap();
- assert!(*test.deleted.clone().lock().await);
+ assert!(*test.stated.clone().lock().await);
}
}
diff --git a/core/src/raw/oio/delete/api.rs b/core/src/raw/oio/delete/api.rs
index 0fa76587d..4c5ec8b2a 100644
--- a/core/src/raw/oio/delete/api.rs
+++ b/core/src/raw/oio/delete/api.rs
@@ -22,31 +22,38 @@ use std::future::Future;
/// Deleter is a type erased [`Delete`]
pub type Deleter = Box<dyn DeleteDyn>;
-/// Delete is the trait to perform delete operations.
+/// The Delete trait defines interfaces for performing deletion operations.
pub trait Delete: Unpin + Send + Sync {
- /// Delete given path with optional arguments.
+ /// Requests deletion of a resource at the specified path with optional arguments
///
- /// # Behavior
+ /// # Parameters
+ /// - `path`: The path of the resource to delete
+ /// - `args`: Additional arguments for the delete operation
///
- /// - `Ok(())` means the path has been queued for deletion.
- /// - `Err(err)` means error happens and no deletion has been queued.
- fn delete(
- &mut self,
- path: String,
- args: OpDelete,
- ) -> impl Future<Output = Result<()>> + MaybeSend;
+ /// # Returns
+ /// - `Ok(())`: The deletion request has been successfully queued (does not guarantee actual deletion)
+ /// - `Err(err)`: An error occurred and the deletion request was not queued
+ ///
+ /// # Notes
+ /// This method just queue the delete request. The actual deletion will be
+ /// performed when `flush` is called.
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()>;
- /// Flush the deletion queue to make sure all enqueued paths are deleted.
+ /// Flushes the deletion queue to ensure queued deletions are executed
///
- /// # Behavior
+ /// # Returns
+ /// - `Ok(0)`: All queued deletions have been processed or the queue is empty.
+ /// - `Ok(count)`: The number of resources successfully deleted. Implementations should
+ /// return an error if the queue is non-empty but no resources were deleted
+ /// - `Err(err)`: An error occurred while performing the deletions
///
- /// - `Ok(())` means all enqueued paths are deleted successfully.
- /// - `Err(err)` means error happens while performing deletion.
- fn flush(&mut self) -> impl Future<Output = Result<()>> + MaybeSend;
+ /// # Notes
+ /// - This method is asynchronous and will wait for queued deletions to complete
+ fn flush(&mut self) -> impl Future<Output = Result<usize>> + MaybeSend;
}
impl Delete for () {
- async fn delete(&mut self, _: String, _: OpDelete) -> Result<()> {
+ fn delete(&mut self, _: String, _: OpDelete) -> Result<()> {
Err(Error::new(
ErrorKind::Unsupported,
"output deleter doesn't support delete",
@@ -62,14 +69,14 @@ impl Delete for () {
}
pub trait DeleteDyn: Unpin + Send + Sync {
- fn delete_dyn(&mut self, path: String, args: OpDelete) -> BoxedFuture<Result<()>>;
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()>;
fn flush_dyn(&mut self) -> BoxedFuture<Result<()>>;
}
impl<T: Delete + ?Sized> DeleteDyn for T {
- fn delete_dyn(&mut self, path: String, args: OpDelete) -> BoxedFuture<Result<()>> {
- Box::pin(self.delete(path, args))
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ Delete::delete(self, path, args)
}
fn flush_dyn(&mut self) -> BoxedFuture<Result<()>> {
@@ -78,8 +85,8 @@ impl<T: Delete + ?Sized> DeleteDyn for T {
}
impl<T: DeleteDyn + ?Sized> Delete for Box<T> {
- async fn delete(&mut self, path: String, args: OpDelete) -> Result<()> {
- self.delete_dyn(path, args).await
+ fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+ DeleteDyn::delete(self, path, args)
}
async fn flush(&mut self) -> Result<()> {
@@ -100,13 +107,14 @@ pub trait BlockingDelete: Send + Sync + 'static {
/// - `Err(err)` means error happens and no deletion has been queued.
fn delete(&mut self, path: &str, args: OpDelete) -> Result<()>;
- /// Flush the deletion queue to make sure all enqueued paths are deleted.
- ///
- /// # Behavior
+ /// Flushes the deletion queue to ensure queued deletions are executed
///
- /// - `Ok(())` means all enqueued paths are deleted successfully.
- /// - `Err(err)` means error happens while performing deletion.
- fn flush(&mut self) -> Result<()>;
+ /// # Returns
+ /// - `Ok(0)`: All queued deletions have been processed or the queue is empty.
+ /// - `Ok(count)`: The number of resources successfully deleted. Implementations should
+ /// return an error if the queue is non-empty but no resources were deleted
+ /// - `Err(err)`: An error occurred while performing the deletions
+ fn flush(&mut self) -> Result<usize>;
}
impl BlockingDelete for () {
@@ -117,7 +125,7 @@ impl BlockingDelete for () {
))
}
- fn flush(&mut self) -> Result<()> {
+ fn flush(&mut self) -> Result<usize> {
Err(Error::new(
ErrorKind::Unsupported,
"output deleter doesn't support flush",
@@ -133,7 +141,7 @@ impl<T: BlockingDelete + ?Sized> BlockingDelete for Box<T> {
(**self).delete(path, args)
}
- fn flush(&mut self) -> Result<()> {
+ fn flush(&mut self) -> Result<usize> {
(**self).flush()
}
}
diff --git a/core/src/raw/oio/delete/batch_delete.rs b/core/src/raw/oio/delete/batch_delete.rs
new file mode 100644
index 000000000..85a4d4852
--- /dev/null
+++ b/core/src/raw/oio/delete/batch_delete.rs
@@ -0,0 +1,125 @@
+// 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::raw::*;
+use crate::*;
+use std::collections::HashSet;
+use std::future::Future;
+
+/// BatchDelete is used to implement [`oio::Delete`] based on batch delete operation.
+///
+/// OneShotDeleter will perform delete operation while calling `flush`.
+pub trait BatchDelete: Send + Sync + Unpin + 'static {
+ /// delete_once delete one path at once.
+ ///
+ /// Implementations should make sure that the data is deleted correctly at once.
+ ///
+ /// BatchDeleter may call this method while there are only one path to delete.
+ fn delete_once(
+ &self,
+ path: String,
+ args: OpDelete,
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
+
+ /// delete_batch delete multiple paths at once.
+ ///
+ /// - Implementations should make sure that the length of `batch` equals to the
+ /// return result's length.
+ /// - Implementations should return error no path is deleted.
+ fn delete_batch(
+ &self,
+ batch: Vec<(String, OpDelete)>,
+ ) -> impl Future<Output = Result<BatchDeleteResult>> + MaybeSend;
+}
+
+/// BatchDeleteResult is the result of batch delete operation.
+#[derive(Default)]
+pub struct BatchDeleteResult {
+ pub succeeded: Vec<(String, OpDelete)>,
+ pub failed: Vec<(String, OpDelete, Error)>,
+}
+
+/// BatchDeleter is used to implement [`oio::Delete`] based on batch delete.
+pub struct BatchDeleter<D: BatchDelete> {
+ inner: D,
+ buffer: HashSet<(String, OpDelete)>,
+}
+
+impl<D: BatchDelete> BatchDeleter<D> {
+ /// Create a new batch deleter.
+ pub fn new(inner: D) -> Self {
+ Self {
+ inner,
+ buffer: HashSet::default(),
+ }
+ }
+}
+
+impl<D: BatchDelete> oio::Delete for BatchDeleter<D> {
+ fn delete(&mut self, path: String, args: OpDelete) -> Result<()> {
+ self.buffer.insert((path, args));
+ Ok(())
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ if self.buffer.is_empty() {
+ return Ok(0);
+ }
+ if self.buffer.len() == 1 {
+ let (path, args) = self
+ .buffer
+ .iter()
+ .next()
+ .expect("the delete buffer size must be 1")
+ .clone();
+ self.inner.delete_once(path, args).await?;
+ self.buffer.clear();
+ return Ok(1);
+ }
+
+ let batch = self.buffer.iter().collect();
+ let mut result = self.inner.delete_batch(batch).await?;
+ debug_assert!(
+ !result.succeeded.is_empty(),
+ "the number of succeeded operations must be greater than 0"
+ );
+ debug_assert_eq!(
+ result.succeeded.len() + result.failed.len(),
+ self.buffer.len(),
+ "the number of succeeded and failed operations must be equal to the input batch size"
+ );
+
+ // Remove all succeeded operations from the buffer.
+ let deleted = result.succeeded.len();
+ for i in result.succeeded {
+ self.buffer.remove(&i);
+ }
+
+ // Return directly if there are non-temporary errors.
+ for (path, op, err) in result.failed {
+ if !err.is_temporary() {
+ return Err(err
+ .with_context("path", path)
+ .with_context("version", op.version()));
+ }
+ }
+
+ // Return the number of succeeded operations to allow users to decide whether
+ // to retry or push more delete operations.
+ Ok(deleted)
+ }
+}
diff --git a/core/src/raw/oio/delete/mod.rs b/core/src/raw/oio/delete/mod.rs
index 75aa0e2de..192a4235b 100644
--- a/core/src/raw/oio/delete/mod.rs
+++ b/core/src/raw/oio/delete/mod.rs
@@ -14,4 +14,20 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
+
mod api;
+pub use api::BlockingDelete;
+pub use api::BlockingDeleter;
+pub use api::Delete;
+pub use api::DeleteDyn;
+pub use api::Deleter;
+
+mod batch_delete;
+pub use batch_delete::BatchDelete;
+pub use batch_delete::BatchDeleteResult;
+pub use batch_delete::BatchDeleter;
+
+mod one_shot_delete;
+pub use one_shot_delete::BlockingOneShotDelete;
+pub use one_shot_delete::OneShotDelete;
+pub use one_shot_delete::OneShotDeleter;
diff --git a/core/src/raw/oio/delete/one_shot_delete.rs b/core/src/raw/oio/delete/one_shot_delete.rs
new file mode 100644
index 000000000..83ddb7be3
--- /dev/null
+++ b/core/src/raw/oio/delete/one_shot_delete.rs
@@ -0,0 +1,104 @@
+// 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::raw::*;
+use crate::*;
+use std::future::Future;
+
+/// OneShotDelete is used to implement [`oio::Delete`] based on one shot operation.
+///
+/// OneShotDeleter will perform delete operation while calling `flush`.
+pub trait OneShotDelete: Send + Sync + Unpin + 'static {
+ /// delete_once delete one path at once.
+ ///
+ /// Implementations should make sure that the data is deleted correctly at once.
+ fn delete_once(
+ &self,
+ path: String,
+ args: OpDelete,
+ ) -> impl Future<Output = Result<()>> + MaybeSend;
+}
+
+/// BlockingOneShotDelete is used to implement [`oio::BlockingDelete`] based on one shot operation.
+///
+/// BlockingOneShotDeleter will perform delete operation while calling `flush`.
+pub trait BlockingOneShotDelete: 'static {
+ /// delete_once delete one path at once.
+ ///
+ /// Implementations should make sure that the data is deleted correctly at once.
+ fn blocking_delete_once(&self, path: String, args: OpDelete) -> Result<()>;
+}
+
+/// OneShotDelete is used to implement [`oio::Delete`] based on one shot.
+pub struct OneShotDeleter<D> {
+ inner: D,
+ delete: Option<(String, OpDelete)>,
+}
+
+impl<D> OneShotDeleter<D> {
+ /// Create a new one shot deleter.
+ pub fn new(inner: D) -> Self {
+ Self {
+ inner,
+ delete: None,
+ }
+ }
+
+ fn delete_inner(&mut self, path: String, args: OpDelete) -> Result<()> {
+ if self.delete.is_some() {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "OneShotDeleter doesn't support batch delete",
+ ));
+ }
+
+ self.delete = Some((path, args));
+ Ok(())
+ }
+}
+
+impl<D: OneShotDelete> oio::Delete for OneShotDeleter<D> {
+ fn delete(&mut self, path: String, args: OpDelete) -> Result<()> {
+ self.delete_inner(path, args)
+ }
+
+ async fn flush(&mut self) -> Result<usize> {
+ let Some((path, args)) = self.delete.clone() else {
+ return Ok(0);
+ };
+
+ self.inner.delete_once(path, args).await?;
+ self.delete = None;
+ Ok(1)
+ }
+}
+
+impl<D: BlockingOneShotDelete> oio::BlockingDelete for OneShotDeleter<D> {
+ fn delete(&mut self, path: String, args: OpDelete) -> Result<()> {
+ self.delete_inner(path, args)
+ }
+
+ fn flush(&mut self) -> Result<usize> {
+ let Some((path, args)) = self.delete.clone() else {
+ return Ok(0);
+ };
+
+ self.inner.blocking_delete_once(path, args)?;
+ self.delete = None;
+ Ok(1)
+ }
+}
diff --git a/core/src/raw/operation.rs b/core/src/raw/operation.rs
index c5ef76047..6f5ce83e1 100644
--- a/core/src/raw/operation.rs
+++ b/core/src/raw/operation.rs
@@ -47,12 +47,14 @@ pub enum Operation {
Stat,
/// Operation for [`crate::raw::Access::delete`]
Delete,
+ /// Operation for [`crate::raw::oio::Delete::delete`]
+ DeleterDelete,
+ /// Operation for [`crate::raw::oio::Delete::flush`]
+ DeleterFlush,
/// Operation for [`crate::raw::Access::list`]
List,
/// Operation for [`crate::raw::oio::List::next`]
ListerNext,
- /// Operation for [`crate::raw::Access::batch`]
- Batch,
/// Operation for [`crate::raw::Access::presign`]
Presign,
/// Operation for [`crate::raw::Access::blocking_create_dir`]
@@ -75,6 +77,10 @@ pub enum Operation {
BlockingStat,
/// Operation for [`crate::raw::Access::blocking_delete`]
BlockingDelete,
+ /// Operation for [`crate::raw::oio::BlockingDelete::delete`]
+ BlockingDeleterDelete,
+ /// Operation for [`crate::raw::oio::BlockingDelete::flush`]
+ BlockingDeleterFlush,
/// Operation for [`crate::raw::Access::blocking_list`]
BlockingList,
/// Operation for [`crate::raw::oio::BlockingList::next`]
@@ -98,9 +104,11 @@ impl Operation {
Operation::ReaderRead
| Operation::WriterWrite
| Operation::ListerNext
+ | Operation::DeleterDelete
| Operation::BlockingReaderRead
| Operation::BlockingWriterWrite
| Operation::BlockingListerNext
+ | Operation::BlockingDeleterDelete
)
}
}
@@ -142,6 +150,10 @@ impl From<Operation> for &'static str {
Operation::BlockingDelete => "blocking_delete",
Operation::BlockingList => "blocking_list",
Operation::BlockingListerNext => "BlockingLister::next",
+ Operation::DeleterDelete => "Deleter::delete",
+ Operation::DeleterFlush => "Deleter::flush",
+ Operation::BlockingDeleterDelete => "BlockingDeleter::delete",
+ Operation::BlockingDeleterFlush => "BlockingDeleter::flush",
}
}
}
diff --git a/core/src/raw/ops.rs b/core/src/raw/ops.rs
index 69615e0c9..ca5895fde 100644
--- a/core/src/raw/ops.rs
+++ b/core/src/raw/ops.rs
@@ -66,6 +66,19 @@ impl OpDelete {
}
}
+/// Args for `delete` operation.
+///
+/// The path must be normalized.
+#[derive(Debug, Clone, Default)]
+pub struct OpDeleter {}
+
+impl OpDeleter {
+ /// Create a new `OpDelete`.
+ pub fn new() -> Self {
+ Self::default()
+ }
+}
+
/// Args for `list` operation.
#[derive(Debug, Clone)]
pub struct OpList {
@@ -247,53 +260,6 @@ impl From<OpWrite> for PresignOperation {
}
}
-/// Args for `batch` operation.
-#[derive(Debug, Clone)]
-pub struct OpBatch {
- ops: Vec<(String, BatchOperation)>,
-}
-
-impl OpBatch {
- /// Create a new batch options.
- pub fn new(ops: Vec<(String, BatchOperation)>) -> Self {
- Self { ops }
- }
-
- /// Get operation from op.
- pub fn operation(&self) -> &[(String, BatchOperation)] {
- &self.ops
- }
-
- /// Consume OpBatch into BatchOperation
- pub fn into_operation(self) -> Vec<(String, BatchOperation)> {
- self.ops
- }
-}
-
-/// Batch operation used for batch.
-#[derive(Debug, Clone)]
-#[non_exhaustive]
-pub enum BatchOperation {
- /// Batch delete operation.
- Delete(OpDelete),
-}
-
-impl From<OpDelete> for BatchOperation {
- fn from(op: OpDelete) -> Self {
- Self::Delete(op)
- }
-}
-
-impl BatchOperation {
- /// Return the operation of this batch.
- pub fn operation(&self) -> Operation {
- use BatchOperation::*;
- match self {
- Delete(_) => Operation::Delete,
- }
- }
-}
-
/// Args for `read` operation.
#[derive(Debug, Clone, Default)]
pub struct OpRead {
diff --git a/core/src/raw/rps.rs b/core/src/raw/rps.rs
index a3073dbd4..62de32879 100644
--- a/core/src/raw/rps.rs
+++ b/core/src/raw/rps.rs
@@ -153,40 +153,6 @@ impl RpRead {
}
}
-/// Reply for `batch` operation.
-pub struct RpBatch {
- results: Vec<(String, Result<BatchedReply>)>,
-}
-
-impl RpBatch {
- /// Create a new RpBatch.
- pub fn new(results: Vec<(String, Result<BatchedReply>)>) -> Self {
- Self { results }
- }
-
- /// Get the results from RpBatch.
- pub fn results(&self) -> &[(String, Result<BatchedReply>)] {
- &self.results
- }
-
- /// Consume RpBatch to get the batched results.
- pub fn into_results(self) -> Vec<(String, Result<BatchedReply>)> {
- self.results
- }
-}
-
-/// Batch results of `batch` operations.
-pub enum BatchedReply {
- /// results of `delete batch` operation
- Delete(RpDelete),
-}
-
-impl From<RpDelete> for BatchedReply {
- fn from(rp: RpDelete) -> Self {
- Self::Delete(rp)
- }
-}
-
/// Reply for `stat` operation.
#[derive(Debug, Clone)]
pub struct RpStat {
diff --git a/core/src/services/aliyun_drive/backend.rs b/core/src/services/aliyun_drive/backend.rs
index 95cf1b754..9e14ba026 100644
--- a/core/src/services/aliyun_drive/backend.rs
+++ b/core/src/services/aliyun_drive/backend.rs
@@ -34,6 +34,7 @@ use super::lister::AliyunDriveLister;
use super::lister::AliyunDriveParent;
use super::writer::AliyunDriveWriter;
use crate::raw::*;
+use crate::services::aliyun_drive::delete::AliyunDriveDeleter;
use crate::services::AliyunDriveConfig;
use crate::*;
@@ -203,9 +204,11 @@ impl Access for AliyunDriveBackend {
type Reader = HttpBody;
type Writer = AliyunDriveWriter;
type Lister = oio::PageLister<AliyunDriveLister>;
+ type Deleter = AliyunDriveDeleter;
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -376,18 +379,11 @@ impl Access for AliyunDriveBackend {
}
}
- async fn delete(&self, path: &str, _args: OpDelete) -> Result<RpDelete> {
- let res = match self.core.get_by_path(path).await {
- Ok(output) => Some(output),
- Err(err) if err.kind() == ErrorKind::NotFound => None,
- Err(err) => return Err(err),
- };
- if let Some(res) = res {
- let file: AliyunDriveFile =
- serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
- self.core.delete_path(&file.file_id).await?;
- }
- Ok(RpDelete::default())
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ AliyunDriveDeleter::new(self.core.clone()),
+ ))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
diff --git a/core/src/services/aliyun_drive/delete.rs b/core/src/services/aliyun_drive/delete.rs
new file mode 100644
index 000000000..0eb9c8be2
--- /dev/null
+++ b/core/src/services/aliyun_drive/delete.rs
@@ -0,0 +1,48 @@
+// 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::core::{AliyunDriveCore, AliyunDriveFile};
+use crate::raw::*;
+use crate::*;
+use bytes::Buf;
+use std::sync::Arc;
+
+pub struct AliyunDriveDeleter {
+ core: Arc<AliyunDriveCore>,
+}
+
+impl AliyunDriveDeleter {
+ pub fn new(core: Arc<AliyunDriveCore>) -> Self {
+ AliyunDriveDeleter { core }
+ }
+}
+
+impl oio::OneShotDelete for AliyunDriveDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let res = match self.core.get_by_path(&path).await {
+ Ok(output) => Some(output),
+ Err(err) if err.kind() == ErrorKind::NotFound => None,
+ Err(err) => return Err(err),
+ };
+ if let Some(res) = res {
+ let file: AliyunDriveFile =
+ serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+ self.core.delete_path(&file.file_id).await?;
+ }
+ Ok(())
+ }
+}
diff --git a/core/src/services/aliyun_drive/mod.rs b/core/src/services/aliyun_drive/mod.rs
index 5481bf4fb..df81a9ee6 100644
--- a/core/src/services/aliyun_drive/mod.rs
+++ b/core/src/services/aliyun_drive/mod.rs
@@ -30,4 +30,6 @@ mod writer;
pub use backend::AliyunDriveBuilder as AliyunDrive;
mod config;
+mod delete;
+
pub use config::AliyunDriveConfig;
diff --git a/core/src/services/alluxio/backend.rs b/core/src/services/alluxio/backend.rs
index a5882c291..4b3a14e7f 100644
--- a/core/src/services/alluxio/backend.rs
+++ b/core/src/services/alluxio/backend.rs
@@ -28,6 +28,7 @@ use super::lister::AlluxioLister;
use super::writer::AlluxioWriter;
use super::writer::AlluxioWriters;
use crate::raw::*;
+use crate::services::alluxio::delete::AlluxioDeleter;
use crate::services::AlluxioConfig;
use crate::*;
@@ -144,9 +145,11 @@ impl Access for AlluxioBackend {
type Reader = HttpBody;
type Writer = AlluxioWriters;
type Lister = oio::PageLister<AlluxioLister>;
+ type Deleter = AlluxioDeleter;
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -204,10 +207,8 @@ impl Access for AlluxioBackend {
Ok((RpWrite::default(), w))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- self.core.delete(path).await?;
-
- Ok(RpDelete::default())
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((RpDelete::default(), AlluxioDeleter::new(self.core.clone())))
}
async fn list(&self, path: &str, _args: OpList) -> Result<(RpList, Self::Lister)> {
diff --git a/core/src/services/azblob/mod.rs b/core/src/services/alluxio/delete.rs
similarity index 66%
copy from core/src/services/azblob/mod.rs
copy to core/src/services/alluxio/delete.rs
index ba65d3439..21837cb92 100644
--- a/core/src/services/azblob/mod.rs
+++ b/core/src/services/alluxio/delete.rs
@@ -15,19 +15,23 @@
// specific language governing permissions and limitations
// under the License.
-#[cfg(feature = "services-azblob")]
-mod core;
-#[cfg(feature = "services-azblob")]
-mod error;
-#[cfg(feature = "services-azblob")]
-mod lister;
-#[cfg(feature = "services-azblob")]
-mod writer;
+use super::core::*;
+use crate::raw::*;
+use crate::*;
+use std::sync::Arc;
-#[cfg(feature = "services-azblob")]
-mod backend;
-#[cfg(feature = "services-azblob")]
-pub use backend::AzblobBuilder as Azblob;
+pub struct AlluxioDeleter {
+ core: Arc<AlluxioCore>,
+}
-mod config;
-pub use config::AzblobConfig;
+impl AlluxioDeleter {
+ pub fn new(core: Arc<AlluxioCore>) -> Self {
+ AlluxioDeleter { core }
+ }
+}
+
+impl oio::OneShotDelete for AlluxioDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ self.core.delete(&path).await
+ }
+}
diff --git a/core/src/services/alluxio/mod.rs b/core/src/services/alluxio/mod.rs
index aa2c88b6f..eac1b3544 100644
--- a/core/src/services/alluxio/mod.rs
+++ b/core/src/services/alluxio/mod.rs
@@ -30,4 +30,6 @@ mod backend;
pub use backend::AlluxioBuilder as Alluxio;
mod config;
+mod delete;
+
pub use config::AlluxioConfig;
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index 638513249..11537428a 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -39,6 +39,7 @@ use super::lister::AzblobLister;
use super::writer::AzblobWriter;
use crate::raw::*;
use crate::services::azblob::core::AzblobCore;
+use crate::services::azblob::delete::AzblobDeleter;
use crate::services::azblob::writer::AzblobWriters;
use crate::services::AzblobConfig;
use crate::*;
@@ -492,9 +493,11 @@ impl Access for AzblobBackend {
type Reader = HttpBody;
type Writer = AzblobWriters;
type Lister = oio::PageLister<AzblobLister>;
+ type Deleter = oio::BatchDeleter<AzblobDeleter>;
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -521,6 +524,8 @@ impl Access for AzblobBackend {
write_with_user_metadata: true,
delete: true,
+ delete_max_size: Some(AZBLOB_BATCH_LIMIT),
+
copy: true,
list: true,
@@ -531,10 +536,6 @@ impl Access for AzblobBackend {
presign_read: self.has_sas_token,
presign_write: self.has_sas_token,
- batch: true,
- batch_delete: true,
- batch_max_operations: Some(self.core.batch_max_operations),
-
..Default::default()
});
@@ -591,15 +592,11 @@ impl Access for AzblobBackend {
Ok((RpWrite::default(), w))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let resp = self.core.azblob_delete_blob(path).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::ACCEPTED | StatusCode::NOT_FOUND => Ok(RpDelete::default()),
- _ => Err(parse_error(resp)),
- }
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::BatchDeleter::new(AzblobDeleter::new(self.core.clone())),
+ ))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -647,77 +644,6 @@ impl Access for AzblobBackend {
parts.headers,
)))
}
-
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let ops = args.into_operation();
- let paths = ops.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
- if paths.len() > AZBLOB_BATCH_LIMIT {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "batch delete limit exceeded",
- ));
- }
-
- // construct and complete batch request
- let resp = self.core.azblob_batch_delete(&paths).await?;
-
- // check response status
- if resp.status() != StatusCode::ACCEPTED {
- return Err(parse_error(resp));
- }
-
- // get boundary from response header
- let content_type = resp.headers().get(CONTENT_TYPE).ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "response data should have CONTENT_TYPE header",
- )
- })?;
- let content_type = content_type
- .to_str()
- .map(|ty| ty.to_string())
- .map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- format!("get invalid CONTENT_TYPE header in response: {:?}", e),
- )
- })?;
- let splits = content_type.split("boundary=").collect::<Vec<&str>>();
- let boundary = splits.get(1).to_owned().ok_or_else(|| {
- Error::new(
- ErrorKind::Unexpected,
- "No boundary message provided in CONTENT_TYPE",
- )
- })?;
-
- let mut bs = resp.into_body();
- let bs = bs.copy_to_bytes(bs.remaining());
-
- let multipart: Multipart<MixedPart> = Multipart::new().with_boundary(boundary).parse(bs)?;
- let parts = multipart.into_parts();
-
- if paths.len() != parts.len() {
- return Err(Error::new(
- ErrorKind::Unexpected,
- "invalid batch response, paths and response parts don't match",
- ));
- }
-
- let mut results = Vec::with_capacity(parts.len());
-
- for (i, part) in parts.into_iter().enumerate() {
- let resp = part.into_response();
- let path = paths[i].clone();
-
- // deleting not existing objects is ok
- if resp.status() == StatusCode::ACCEPTED || resp.status() == StatusCode::NOT_FOUND {
- results.push((path, Ok(RpDelete::default().into())));
- } else {
- results.push((path, Err(parse_error(resp))));
- }
- }
- Ok(RpBatch::new(results))
- }
}
#[cfg(test)]
diff --git a/core/src/services/azblob/delete.rs b/core/src/services/azblob/delete.rs
new file mode 100644
index 000000000..252d220dc
--- /dev/null
+++ b/core/src/services/azblob/delete.rs
@@ -0,0 +1,103 @@
+// 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::core::*;
+use super::error::parse_error;
+use crate::raw::oio::BatchDeleteResult;
+use crate::raw::*;
+use crate::*;
+use http::StatusCode;
+use std::sync::Arc;
+
+pub struct AzblobDeleter {
+ core: Arc<AzblobCore>,
+}
+
+impl AzblobDeleter {
+ pub fn new(core: Arc<AzblobCore>) -> Self {
+ Self { core }
+ }
+}
+
+impl oio::BatchDelete for AzblobDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let resp = self.core.azblob_delete_blob(&path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::ACCEPTED | StatusCode::NOT_FOUND => Ok(()),
+ _ => Err(parse_error(resp)),
+ }
+ }
+
+ async fn delete_batch(&self, batch: Vec<(String, OpDelete)>) -> Result<BatchDeleteResult> {
+ // TODO: Add remove version support.
+ let paths = batch.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
+
+ // construct and complete batch request
+ let resp = self.core.azblob_batch_delete(&paths).await?;
+
+ // check response status
+ if resp.status() != StatusCode::ACCEPTED {
+ return Err(parse_error(resp));
+ }
+
+ // get boundary from response header
+ let boundary = parse_multipart_boundary(resp.headers()).ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "invalid response: no boundary provided in header",
+ )
+ })?;
+
+ let bs = resp.into_body().to_bytes();
+ let multipart: Multipart<MixedPart> = Multipart::new().with_boundary(boundary).parse(bs)?;
+ let parts = multipart.into_parts();
+
+ if paths.len() != parts.len() {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "invalid batch response, paths and response parts don't match",
+ ));
+ }
+
+ let mut batched_result = BatchDeleteResult::default();
+
+ for (i, part) in parts.into_iter().enumerate() {
+ let resp = part.into_response();
+ let path = paths[i].clone();
+
+ // deleting not existing objects is ok
+ if resp.status() == StatusCode::ACCEPTED || resp.status() == StatusCode::NOT_FOUND {
+ batched_result.succeeded.push((path, OpDelete::default()));
+ } else {
+ batched_result
+ .failed
+ .push((path, OpDelete::default(), parse_error(resp)));
+ }
+ }
+
+ // If no object is deleted, return directly.
+ if batched_result.succeeded.is_empty() {
+ let err = batched_result.failed.remove(0).2;
+ return Err(err);
+ }
+
+ Ok(batched_result)
+ }
+}
diff --git a/core/src/services/azblob/mod.rs b/core/src/services/azblob/mod.rs
index ba65d3439..c78536379 100644
--- a/core/src/services/azblob/mod.rs
+++ b/core/src/services/azblob/mod.rs
@@ -30,4 +30,6 @@ mod backend;
pub use backend::AzblobBuilder as Azblob;
mod config;
+mod delete;
+
pub use config::AzblobConfig;
diff --git a/core/src/services/azdls/backend.rs b/core/src/services/azdls/backend.rs
index 34d7125d6..ce2ccadd7 100644
--- a/core/src/services/azdls/backend.rs
+++ b/core/src/services/azdls/backend.rs
@@ -32,6 +32,7 @@ use super::lister::AzdlsLister;
use super::writer::AzdlsWriter;
use super::writer::AzdlsWriters;
use crate::raw::*;
+use crate::services::azdls::delete::AzdlsDeleter;
use crate::services::AzdlsConfig;
use crate::*;
@@ -217,9 +218,11 @@ impl Access for AzdlsBackend {
type Reader = HttpBody;
type Writer = AzdlsWriters;
type Lister = oio::PageLister<AzdlsLister>;
+ type Deleter = oio::OneShotDeleter<AzdlsDeleter>;
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -335,15 +338,11 @@ impl Access for AzdlsBackend {
Ok((RpWrite::default(), w))
}
- async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
- let resp = self.core.azdls_delete(path).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK | StatusCode::NOT_FOUND => Ok(RpDelete::default()),
- _ => Err(parse_error(resp)),
- }
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((
+ RpDelete::default(),
+ oio::OneShotDeleter::new(AzdlsDeleter::new(self.core.clone())),
+ ))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
diff --git a/core/src/services/azblob/mod.rs b/core/src/services/azdls/delete.rs
similarity index 55%
copy from core/src/services/azblob/mod.rs
copy to core/src/services/azdls/delete.rs
index ba65d3439..3fe7a557c 100644
--- a/core/src/services/azblob/mod.rs
+++ b/core/src/services/azdls/delete.rs
@@ -15,19 +15,32 @@
// specific language governing permissions and limitations
// under the License.
-#[cfg(feature = "services-azblob")]
-mod core;
-#[cfg(feature = "services-azblob")]
-mod error;
-#[cfg(feature = "services-azblob")]
-mod lister;
-#[cfg(feature = "services-azblob")]
-mod writer;
+use super::core::*;
+use super::error::parse_error;
+use crate::raw::*;
+use crate::*;
+use http::StatusCode;
+use std::sync::Arc;
-#[cfg(feature = "services-azblob")]
-mod backend;
-#[cfg(feature = "services-azblob")]
-pub use backend::AzblobBuilder as Azblob;
+pub struct AzdlsDeleter {
+ core: Arc<AzdlsCore>,
+}
-mod config;
-pub use config::AzblobConfig;
+impl AzdlsDeleter {
+ pub fn new(core: Arc<AzdlsCore>) -> Self {
+ Self { core }
+ }
+}
+
+impl oio::OneShotDelete for AzdlsDeleter {
+ async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
+ let resp = self.core.azdls_delete(&path).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK | StatusCode::NOT_FOUND => Ok(()),
+ _ => Err(parse_error(resp)),
+ }
+ }
+}
diff --git a/core/src/services/azdls/mod.rs b/core/src/services/azdls/mod.rs
index 321899ccc..788aa2351 100644
--- a/core/src/services/azdls/mod.rs
+++ b/core/src/services/azdls/mod.rs
@@ -30,4 +30,6 @@ mod backend;
pub use backend::AzdlsBuilder as Azdls;
mod config;
+mod delete;
+
pub use config::AzdlsConfig;
diff --git a/core/src/services/dropbox/backend.rs b/core/src/services/dropbox/backend.rs
index 9cc96b0dd..71106f75e 100644
--- a/core/src/services/dropbox/backend.rs
+++ b/core/src/services/dropbox/backend.rs
@@ -65,9 +65,6 @@ impl Access for DropboxBackend {
rename: true,
- batch: true,
- batch_delete: true,
-
..Default::default()
});
ma.into()
@@ -231,52 +228,4 @@ impl Access for DropboxBackend {
}
}
}
-
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let ops = args.into_operation();
- if ops.len() > 1000 {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "dropbox services only allow delete up to 1000 keys at once",
- )
- .with_context("length", ops.len().to_string()));
- }
-
- let paths = ops.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
-
- let resp = self.core.dropbox_delete_batch(paths).await?;
- if resp.status() != StatusCode::OK {
- return Err(parse_error(resp));
- }
-
- let bs = resp.into_body();
- let decoded_response: DropboxDeleteBatchResponse =
- serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
-
- match decoded_response.tag.as_str() {
- "complete" => {
- let entries = decoded_response.entries.unwrap_or_default();
- let results = self.core.handle_batch_delete_complete_result(entries);
- Ok(RpBatch::new(results))
- }
- "async_job_id" => {
- let job_id = decoded_response
- .async_job_id
- .expect("async_job_id should be present");
- let res = { || self.core.dropbox_delete_batch_check(job_id.clone()) }
- .retry(*BACKOFF)
- .when(|e| e.is_temporary())
- .await?;
-
- Ok(res)
- }
- _ => Err(Error::new(
- ErrorKind::Unexpected,
- format!(
- "delete batch failed with unexpected tag {}",
- decoded_response.tag
- ),
- )),
- }
- }
}
diff --git a/core/src/services/dropbox/core.rs b/core/src/services/dropbox/core.rs
index 1162eb5e7..1867175ea 100644
--- a/core/src/services/dropbox/core.rs
+++ b/core/src/services/dropbox/core.rs
@@ -42,14 +42,6 @@ use super::error::parse_error;
use crate::raw::*;
use crate::*;
-/// BACKOFF is the backoff used inside dropbox to make sure dropbox async task succeed.
-pub static BACKOFF: Lazy<ExponentialBuilder> = Lazy::new(|| {
- ExponentialBuilder::default()
- .with_max_delay(Duration::from_secs(10))
- .with_max_times(10)
- .with_jitter()
-});
-
pub struct DropboxCore {
pub root: String,
@@ -202,73 +194,6 @@ impl DropboxCore {
self.client.send(request).await
}
- pub async fn dropbox_delete_batch(&self, paths: Vec<String>) -> Result<Response<Buffer>> {
- let url = "https://api.dropboxapi.com/2/files/delete_batch".to_string();
- let args = DropboxDeleteBatchArgs {
- entries: paths
- .into_iter()
- .map(|path| DropboxDeleteBatchEntry {
- path: self.build_path(&path),
- })
- .collect(),
- };
-
- let bs = Bytes::from(serde_json::to_string(&args).map_err(new_json_serialize_error)?);
-
- let mut request = Request::post(&url)
- .header(CONTENT_TYPE, "application/json")
- .header(CONTENT_LENGTH, bs.len())
- .body(Buffer::from(bs))
- .map_err(new_request_build_error)?;
-
- self.sign(&mut request).await?;
- self.client.send(request).await
- }
-
- pub async fn dropbox_delete_batch_check(&self, async_job_id: String) -> Result<RpBatch> {
- let url = "https://api.dropboxapi.com/2/files/delete_batch/check".to_string();
- let args = DropboxDeleteBatchCheckArgs { async_job_id };
-
- let bs = Bytes::from(serde_json::to_vec(&args).map_err(new_json_serialize_error)?);
-
- let mut request = Request::post(&url)
- .header(CONTENT_TYPE, "application/json")
- .header(CONTENT_LENGTH, bs.len())
- .body(Buffer::from(bs))
- .map_err(new_request_build_error)?;
-
- self.sign(&mut request).await?;
-
- let resp = self.client.send(request).await?;
- if resp.status() != StatusCode::OK {
- return Err(parse_error(resp));
- }
-
- let bs = resp.into_body();
-
- let decoded_response: DropboxDeleteBatchResponse =
- serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
- match decoded_response.tag.as_str() {
- "in_progress" => Err(Error::new(
- ErrorKind::Unexpected,
- "delete batch job still in progress",
- )
- .set_temporary()),
- "complete" => {
- let entries = decoded_response.entries.unwrap_or_default();
- let results = self.handle_batch_delete_complete_result(entries);
- Ok(RpBatch::new(results))
- }
- _ => Err(Error::new(
- ErrorKind::Unexpected,
- format!(
- "delete batch check failed with unexpected tag {}",
- decoded_response.tag
- ),
- )),
- }
- }
-
pub async fn dropbox_create_folder(&self, path: &str) -> Result<RpCreateDir> {
let url = "https://api.dropboxapi.com/2/files/create_folder_v2".to_string();
let args = DropboxCreateFolderArgs {
@@ -404,53 +329,6 @@ impl DropboxCore {
self.client.send(request).await
}
-
- pub fn handle_batch_delete_complete_result(
- &self,
- entries: Vec<DropboxDeleteBatchResponseEntry>,
- ) -> Vec<(String, Result<BatchedReply>)> {
- let mut results = Vec::with_capacity(entries.len());
- for entry in entries {
- let result = match entry.tag.as_str() {
- // Only success response has metadata and then path,
- // so we cannot tell which path failed.
- "success" => {
- let path = entry
- .metadata
- .expect("metadata should be present")
- .path_display;
- (path, Ok(RpDelete::default().into()))
- }
- "failure" => {
- let error = entry.failure.expect("error should be present");
- let error_cause = &error
- .failure_cause_map
- .get(&error.tag)
- .expect("error should be present")
- .tag;
- // Ignore errors about path lookup not found and report others.
- if error.tag == "path_lookup" && error_cause == "not_found" {
- ("".to_string(), Ok(RpDelete::default().into()))
- } else {
- let err = Error::new(
- ErrorKind::Unexpected,
- format!("delete failed with error {} {}", error.tag, error_cause),
- );
- ("".to_string(), Err(err))
- }
- }
- _ => (
- "".to_string(),
- Err(Error::new(
- ErrorKind::Unexpected,
- format!("delete failed with unexpected tag {}", entry.tag),
- )),
- ),
- };
- results.push(result);
- }
- results
- }
}
#[derive(Clone)]
diff --git a/core/src/services/lakefs/backend.rs b/core/src/services/lakefs/backend.rs
index dc913e252..b6dd701a1 100644
--- a/core/src/services/lakefs/backend.rs
+++ b/core/src/services/lakefs/backend.rs
@@ -288,7 +288,7 @@ impl Access for LakefsBackend {
))
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
// This would delete the bucket, do not perform
if self.core.root == "/" && path == "/" {
return Ok(RpDelete::default());
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index a627e1d4a..83e902173 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -534,7 +534,7 @@ impl Access for OssBackend {
Ok((RpWrite::default(), w))
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
let resp = self.core.oss_delete_object(path, &args).await?;
let status = resp.status();
match status {
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 5b523fc72..f876e5f29 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -42,6 +42,7 @@ use reqsign::AwsV4Signer;
use reqwest::Url;
use super::core::*;
+use super::delete::S3Deleter;
use super::error::parse_error;
use super::error::parse_s3_error_code;
use super::lister::{S3Lister, S3Listers, S3ObjectVersionsLister};
@@ -893,9 +894,11 @@ impl Access for S3Backend {
type Reader = HttpBody;
type Writer = S3Writers;
type Lister = S3Listers;
+ type Deleter = S3Deleter;
type BlockingReader = ();
type BlockingWriter = ();
type BlockingLister = ();
+ type BlockingDeleter = ();
fn info(&self) -> Arc<AccessorInfo> {
let mut am = AccessorInfo::default();
@@ -1016,24 +1019,8 @@ impl Access for S3Backend {
Ok((RpWrite::default(), w))
}
- async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
- // This would delete the bucket, do not perform
- if self.core.root == "/" && path == "/" {
- return Ok(RpDelete::default());
- }
-
- let resp = self.core.s3_delete_object(path, &args).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::NO_CONTENT => Ok(RpDelete::default()),
- // Allow 404 when deleting a non-existing object
- // This is not a standard behavior, only some s3 alike service like GCS XML API do this.
- // ref: <https://cloud.google.com/storage/docs/xml-api/delete-object>
- StatusCode::NOT_FOUND => Ok(RpDelete::default()),
- _ => Err(parse_error(resp)),
- }
+ async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
+ Ok((RpDelete::default(), S3Deleter::new(self.core.clone())))
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
@@ -1096,53 +1083,6 @@ impl Access for S3Backend {
parts.headers,
)))
}
-
- async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let ops = args.into_operation();
- if ops.len() > 1000 {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "s3 services only allow delete up to 1000 keys at once",
- )
- .with_context("length", ops.len().to_string()));
- }
-
- let paths = ops.into_iter().map(|(p, _)| p).collect();
-
- let resp = self.core.s3_delete_objects(paths).await?;
-
- let status = resp.status();
-
- if let StatusCode::OK = status {
- let bs = resp.into_body();
-
- let result: DeleteObjectsResult =
- quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
-
- let mut batched_result = Vec::with_capacity(result.deleted.len() + result.error.len());
- for i in result.deleted {
- let path = build_rel_path(&self.core.root, &i.key);
- batched_result.push((path, Ok(RpDelete::default().into())));
- }
- for i in result.error {
- let path = build_rel_path(&self.core.root, &i.key);
-
- // set the error kind and mark temporary if retryable
- let (kind, retryable) =
- parse_s3_error_code(i.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
- let mut err: Error = Error::new(kind, format!("{i:?}"));
- if retryable {
- err = err.set_temporary();
- }
-
- batched_result.push((path, Err(err)));
- }
-
- Ok(RpBatch::new(batched_result))
- } else {
- Err(parse_error(resp))
- }
- }
}
#[cfg(test)]
diff --git a/core/src/services/s3/delete.rs b/core/src/services/s3/delete.rs
new file mode 100644
index 000000000..c51c77d8b
--- /dev/null
+++ b/core/src/services/s3/delete.rs
@@ -0,0 +1,111 @@
+// 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::core::*;
+use super::error::{parse_error, parse_s3_error_code};
+use crate::raw::oio::BatchDeleteResult;
+use crate::raw::*;
+use crate::*;
+use bytes::Buf;
+use http::StatusCode;
+use std::sync::Arc;
+
+pub struct S3Deleter {
+ core: Arc<S3Core>,
+}
+
+impl S3Deleter {
+ pub fn new(core: Arc<S3Core>) -> Self {
+ Self { core }
+ }
+}
+
+impl oio::BatchDelete for S3Deleter {
+ async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+ // This would delete the bucket, do not perform
+ if self.core.root == "/" && path == "/" {
+ return Ok(());
+ }
+
+ let resp = self.core.s3_delete_object(&path, &args).await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::NO_CONTENT => Ok(()),
+ // Allow 404 when deleting a non-existing object
+ // This is not a standard behavior, only some s3 alike service like GCS XML API do this.
+ // ref: <https://cloud.google.com/storage/docs/xml-api/delete-object>
+ StatusCode::NOT_FOUND => Ok(()),
+ _ => Err(parse_error(resp)),
+ }
+ }
+
+ async fn delete_batch(&self, batch: Vec<(String, OpDelete)>) -> Result<BatchDeleteResult> {
+ // TODO: fix https://github.com/apache/opendal/issues/5329
+ let paths = batch.into_iter().map(|(p, _)| p).collect();
+
+ let resp = self.core.s3_delete_objects(paths).await?;
+
+ let status = resp.status();
+ if status != StatusCode::OK {
+ return Err(parse_error(resp));
+ }
+
+ let bs = resp.into_body();
+
+ let mut result: DeleteObjectsResult =
+ quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
+
+ // If no object is deleted, return directly.
+ if result.deleted.is_empty() {
+ let err = result.error.remove(0);
+ return Err(parse_delete_objects_result_error(err));
+ }
+
+ let mut batched_result = BatchDeleteResult {
+ succeeded: Vec::with_capacity(result.deleted.len()),
+ failed: Vec::with_capacity(result.error.len()),
+ };
+ for i in result.deleted {
+ let path = build_rel_path(&self.core.root, &i.key);
+ // TODO: fix https://github.com/apache/opendal/issues/5329
+ batched_result.succeeded.push((path, OpDelete::new()));
+ }
+ for i in result.error {
+ let path = build_rel_path(&self.core.root, &i.key);
+
+ batched_result.failed.push((
+ path,
+ OpDelete::new(),
+ parse_delete_objects_result_error(i),
+ ));
+ }
+
+ Ok(batched_result)
+ }
+}
+
+fn parse_delete_objects_result_error(err: DeleteObjectsResultError) -> Error {
+ let (kind, retryable) =
+ parse_s3_error_code(err.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
+ let mut err: Error = Error::new(kind, format!("{err:?}"));
+ if retryable {
+ err = err.set_temporary();
+ }
+ err
+}
diff --git a/core/src/services/s3/mod.rs b/core/src/services/s3/mod.rs
index b4161a9d6..2e13b4e3a 100644
--- a/core/src/services/s3/mod.rs
+++ b/core/src/services/s3/mod.rs
@@ -30,4 +30,6 @@ mod backend;
pub use backend::S3Builder as S3;
mod config;
+mod delete;
+
pub use config::S3Config;
diff --git a/core/src/types/capability.rs b/core/src/types/capability.rs
index b30cbf880..fd3191de9 100644
--- a/core/src/types/capability.rs
+++ b/core/src/types/capability.rs
@@ -151,6 +151,8 @@ pub struct Capability {
pub delete: bool,
/// Indicates if versioned delete operations are supported.
pub delete_with_version: bool,
+ /// Maximum size supported for single delete operations.
+ pub delete_max_size: Option<usize>,
/// Indicates if copy operations are supported.
pub copy: bool,
diff --git a/core/src/types/delete.rs b/core/src/types/delete.rs
new file mode 100644
index 000000000..be9eef4bb
--- /dev/null
+++ b/core/src/types/delete.rs
@@ -0,0 +1,197 @@
+// 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::raw::oio::Delete;
+use crate::raw::*;
+use crate::*;
+use futures::Sink;
+use std::pin::Pin;
+use std::task::{ready, Context, Poll};
+
+pub struct Deleter {
+ state: State,
+
+ max_size: usize,
+ cur_size: usize,
+}
+
+enum State {
+ Idle(Option<oio::Deleter>),
+ Flush(BoxedStaticFuture<(oio::Deleter, Result<usize>)>),
+}
+
+/// # Safety
+///
+/// Lister will only be accessed by `&mut Self`
+unsafe impl Sync for Deleter {}
+
+impl Deleter {
+ /// Create a new lister.
+ pub(crate) async fn create(acc: Accessor) -> Result<Self> {
+ let max_size = acc.info().full_capability().delete_max_size.unwrap_or(1);
+ let (_, deleter) = acc.delete().await?;
+
+ Ok(Self {
+ state: State::Idle(Some(deleter)),
+ max_size,
+ cur_size: 0,
+ })
+ }
+
+ fn poll_ready_inner(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ loop {
+ match &mut self.state {
+ State::Idle(deleter) => {
+ if self.cur_size < self.max_size {
+ return Poll::Ready(Ok(()));
+ } else {
+ let mut deleter = deleter
+ .take()
+ .expect("deleter must be valid while calling poll_ready");
+ let fut = async move {
+ let res = deleter.flush().await;
+ (deleter, res)
+ };
+ self.state = State::Flush(BoxedStaticFuture::new(fut));
+ }
+ }
+ State::Flush(fut) => {
+ let (deleter, res) = ready!(fut.as_mut().poll(cx));
+ self.state = State::Idle(Some(deleter));
+ return Poll::Ready(res.map(|flushed| self.cur_size -= flushed));
+ }
+ }
+ }
+ }
+
+ fn poll_flush_inner(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ loop {
+ match &mut self.state {
+ // We allow users to call flush even if the max size is not reached.
+ State::Idle(deleter) => {
+ let mut deleter = deleter
+ .take()
+ .expect("deleter must be valid while calling poll_ready");
+ let fut = async move {
+ let res = deleter.flush().await;
+ (deleter, res)
+ };
+ self.state = State::Flush(BoxedStaticFuture::new(fut));
+ }
+ State::Flush(fut) => {
+ let (deleter, res) = ready!(fut.as_mut().poll(cx));
+ self.state = State::Idle(Some(deleter));
+ return Poll::Ready(res.map(|flushed| self.cur_size -= flushed));
+ }
+ }
+ }
+ }
+
+ fn poll_close_inner(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ loop {
+ match &mut self.state {
+ State::Idle(deleter) => {
+ let mut deleter = deleter
+ .take()
+ .expect("deleter must be valid while calling poll_ready");
+ let fut = async move {
+ let res = deleter.flush().await;
+ (deleter, res)
+ };
+ self.state = State::Flush(BoxedStaticFuture::new(fut));
+ }
+ State::Flush(fut) => {
+ let (deleter, res) = ready!(fut.as_mut().poll(cx));
+ self.state = State::Idle(Some(deleter));
+ self.cur_size -= res?;
+ // Keep looping if there are still pending delete operations.
+ if self.cur_size == 0 {
+ return Poll::Ready(Ok(()));
+ }
+ }
+ }
+ }
+ }
+}
+
+impl Sink<String> for Deleter {
+ type Error = Error;
+
+ fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_ready_inner(cx)
+ }
+
+ fn start_send(self: Pin<&mut Self>, item: String) -> Result<()> {
+ match &mut self.state {
+ State::Idle(deleter) if self.cur_size < self.max_size => {
+ let mut deleter = deleter
+ .take()
+ .expect("deleter must be valid while calling start_send");
+ let res = deleter.delete(item, OpDelete::default());
+ self.state = State::Idle(Some(deleter));
+ self.cur_size += 1;
+ res
+ }
+ _ => Err(Error::new(
+ ErrorKind::Unexpected,
+ "Deleter is not ready to start_send, please poll_ready first",
+ )),
+ }
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_flush_inner(cx)
+ }
+
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_close_inner(cx)
+ }
+}
+
+impl Sink<(String, OpDelete)> for Deleter {
+ type Error = Error;
+
+ fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_ready_inner(cx)
+ }
+
+ fn start_send(self: Pin<&mut Self>, item: (String, OpDelete)) -> Result<()> {
+ match &mut self.state {
+ State::Idle(deleter) if self.cur_size < self.max_size => {
+ let mut deleter = deleter
+ .take()
+ .expect("deleter must be valid while calling start_send");
+ let res = deleter.delete(item.0, item.1);
+ self.state = State::Idle(Some(deleter));
+ self.cur_size += 1;
+ res
+ }
+ _ => Err(Error::new(
+ ErrorKind::Unexpected,
+ "Deleter is not ready to start_send, please poll_ready first",
+ )),
+ }
+ }
+
+ fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_flush_inner(cx)
+ }
+
+ fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<()>> {
+ self.poll_close_inner(cx)
+ }
+}
diff --git a/core/src/types/mod.rs b/core/src/types/mod.rs
index 448a5f430..9e4dcf3ab 100644
--- a/core/src/types/mod.rs
+++ b/core/src/types/mod.rs
@@ -70,4 +70,6 @@ mod capability;
pub use capability::Capability;
mod context;
+mod delete;
+
pub(crate) use context::*;
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 0e8af0298..96d7372ea 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -18,14 +18,16 @@
use std::future::Future;
use std::time::Duration;
-use futures::stream;
use futures::Stream;
use futures::StreamExt;
use futures::TryStreamExt;
+use futures::{stream, SinkExt};
use super::BlockingOperator;
use crate::operator_futures::*;
+use crate::raw::oio::Delete;
use crate::raw::*;
+use crate::types::delete::Deleter;
use crate::*;
/// Operator is the entry for all public async APIs.
@@ -1501,12 +1503,19 @@ impl Operator {
path,
OpDelete::default(),
|inner, path, args| async move {
- let _ = inner.delete(&path, args).await?;
+ let (_, mut deleter) = inner.delete().await?;
+ deleter.delete(path, args).await?;
+ deleter.flush().await?;
Ok(())
},
)
}
+ /// TODO.
+ pub async fn deleter(&self) -> Result<Deleter> {
+ Deleter::create(self.inner().clone()).await
+ }
+
///
/// # Notes
///
@@ -1526,8 +1535,14 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
+ #[deprecated("use `Deleter::send_all` instead")]
pub async fn remove(&self, paths: Vec<String>) -> Result<()> {
- self.remove_via(stream::iter(paths)).await
+ let mut deleter = self.deleter().await?;
+ deleter
+ .send_all(&mut stream::iter(paths).map(|v| Ok(normalize_path(&v))))
+ .await?;
+ deleter.close().await?;
+ Ok(())
}
/// remove will remove files via the given paths.
@@ -1555,35 +1570,13 @@ impl Operator {
/// # Ok(())
/// # }
/// ```
+ #[deprecated("use `Deleter::send_all` instead")]
pub async fn remove_via(&self, input: impl Stream<Item = String> + Unpin) -> Result<()> {
- let input = input.map(|v| normalize_path(&v));
-
- if self.info().full_capability().batch {
- let mut input = input
- .map(|v| (v, OpDelete::default().into()))
- .chunks(self.limit());
-
- while let Some(batches) = input.next().await {
- let results = self
- .inner()
- .batch(OpBatch::new(batches))
- .await?
- .into_results();
-
- // TODO: return error here directly seems not a good idea?
- for (_, result) in results {
- let _ = result?;
- }
- }
- } else {
- input
- .map(Ok)
- .try_for_each_concurrent(self.limit, |path| async move {
- let _ = self.inner().delete(&path, OpDelete::default()).await?;
- Ok::<(), Error>(())
- })
- .await?;
- }
+ let mut deleter = self.deleter().await?;
+ deleter
+ .send_all(&mut input.map(|v| Ok(normalize_path(&v))))
+ .await?;
+ deleter.close().await?;
Ok(())
}
@@ -1626,33 +1619,10 @@ impl Operator {
Err(e) => return Err(e),
};
- let obs = self.lister_with(path).recursive(true).await?;
-
- if self.info().full_capability().batch {
- let mut obs = obs.try_chunks(self.limit());
-
- while let Some(batches) = obs.next().await {
- let batches = batches
- .map_err(|err| err.1)?
- .into_iter()
- .map(|v| (v.path().to_string(), OpDelete::default().into()))
- .collect();
-
- let results = self
- .inner()
- .batch(OpBatch::new(batches))
- .await?
- .into_results();
-
- // TODO: return error here directly seems not a good idea?
- for (_, result) in results {
- let _ = result?;
- }
- }
- } else {
- obs.try_for_each(|v| async move { self.delete(v.path()).await })
- .await?;
- }
+ let mut lister = self.lister_with(path).recursive(true).await?;
+ let mut deleter = self.deleter().await?;
+ deleter.send_all(&mut lister).await?;
+ deleter.close().await?;
Ok(())
}
diff --git a/core/src/types/operator/operator_futures.rs b/core/src/types/operator/operator_futures.rs
index 53d89f594..23cb86d1b 100644
--- a/core/src/types/operator/operator_futures.rs
+++ b/core/src/types/operator/operator_futures.rs
@@ -434,6 +434,11 @@ impl<F: Future<Output = Result<()>>> FutureDelete<F> {
}
}
+/// Future that generated by [`Operator::deleter_with`].
+///
+/// Users can add more options by public functions provided by this struct.
+pub type FutureDeleter<F> = OperatorFuture<OpDeleter, (), F>;
+
/// Future that generated by [`Operator::list_with`] or [`Operator::lister_with`].
///
/// Users can add more options by public functions provided by this struct.