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 2023/04/02 17:44:58 UTC
[incubator-opendal] 01/01: refactor(core): Simplify the usage of BatchOperation and BatchResults
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch polish-retry
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit d5e0ccaae6feb6ef5278c40eb75c4bb3256bec73
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Apr 3 01:44:18 2023 +0800
refactor(core): Simplify the usage of BatchOperation and BatchResults
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
core/src/layers/error_context.rs | 7 +--
core/src/layers/logging.rs | 6 +-
core/src/layers/retry.rs | 107 ++++++++++++++++++++----------------
core/src/raw/rps.rs | 54 +++---------------
core/src/services/azblob/backend.rs | 97 ++++++++++++++++----------------
core/src/services/oss/backend.rs | 104 +++++++++++++++++------------------
core/src/services/s3/backend.rs | 69 +++++++++++------------
core/src/types/operator/operator.rs | 20 +++----
core/src/types/ops.rs | 40 +++++---------
9 files changed, 231 insertions(+), 273 deletions(-)
diff --git a/core/src/layers/error_context.rs b/core/src/layers/error_context.rs
index a5d58cfc..2546c5c4 100644
--- a/core/src/layers/error_context.rs
+++ b/core/src/layers/error_context.rs
@@ -214,9 +214,8 @@ impl<A: Accessor> LayeredAccessor for ErrorContextAccessor<A> {
self.inner
.batch(args)
.map_ok(|v| {
- let BatchedResults::Delete(res) = v.into_results();
-
- let res = res
+ let res = v
+ .into_results()
.into_iter()
.map(|(path, res)| {
let res = res.map_err(|err| {
@@ -228,7 +227,7 @@ impl<A: Accessor> LayeredAccessor for ErrorContextAccessor<A> {
})
.collect();
- RpBatch::new(BatchedResults::Delete(res))
+ RpBatch::new(res)
})
.map_err(|err| {
err.with_operation(Operation::Batch)
diff --git a/core/src/layers/logging.rs b/core/src/layers/logging.rs
index 69712e6e..923ace20 100644
--- a/core/src/layers/logging.rs
+++ b/core/src/layers/logging.rs
@@ -532,7 +532,7 @@ impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
}
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
- let (op, count) = (args.operation().operation(), args.operation().len());
+ let (op, count) = (args.operation()[0].1.operation(), args.operation().len());
debug!(
target: LOGGING_TARGET,
@@ -550,8 +550,8 @@ impl<A: Accessor> LayeredAccessor for LoggingAccessor<A> {
self.scheme,
Operation::Batch,
v.results().len(),
- v.results().len_ok(),
- v.results().len_err(),
+ v.results().iter().filter(|(_, v)|v.is_ok()).count(),
+ v.results().iter().filter(|(_, v)|v.is_err()).count(),
);
v
})
diff --git a/core/src/layers/retry.rs b/core/src/layers/retry.rs
index 3e251a0a..359d59ce 100644
--- a/core/src/layers/retry.rs
+++ b/core/src/layers/retry.rs
@@ -291,7 +291,13 @@ impl<A: Accessor> LayeredAccessor for RetryAccessor<A> {
{ || self.inner.batch(args.clone()).map(Err::<(), _>) }
.retry(&self.builder)
.when(|res| match res {
- Ok(rp) => rp.results().errors().any(|e| e.is_temporary()),
+ Ok(rp) => rp.results().iter().any(|(_, v)| {
+ if let Err(e) = v {
+ e.is_temporary()
+ } else {
+ false
+ }
+ }),
Err(err) => err.is_temporary(),
})
.notify(|res, dur| match res {
@@ -299,7 +305,14 @@ impl<A: Accessor> LayeredAccessor for RetryAccessor<A> {
target: "opendal::service",
"operation={} -> retry after {}s: error={:?}",
Operation::Batch, dur.as_secs_f64(),
- rp.results().errors().filter(|e| e.is_temporary()).collect::<Vec<_>>()),
+ rp.results().iter().filter_map(|(_, v)| {
+ if let Err(e) = v {
+ if e.is_temporary() {
+ return Some(e);
+ }
+ }
+ None
+ }).collect::<Vec<_>>()),
Err(err) => warn!(
target: "opendal::service",
"operation={} -> retry after {}s: error={:?}",
@@ -842,54 +855,56 @@ mod tests {
let mut attempt = self.attempt.lock().unwrap();
*attempt += 1;
- use BatchOperations::*;
- match op.into_operation() {
- Delete(v) => match *attempt {
- 1 => Err(
- Error::new(ErrorKind::Unexpected, "retryable_error from reader")
- .set_temporary(),
- ),
- 2 => Ok(RpBatch::new(BatchedResults::Delete(
- v.into_iter()
- .map(|(s, _)| {
- (
- s,
- Err(Error::new(
+ 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()),
- )
- })
- .collect(),
- ))),
- 3 => Ok(RpBatch::new(BatchedResults::Delete(
- v.into_iter()
- .enumerate()
- .map(|(i, (s, _))| {
- (
- s,
- match i {
- 0 => Err(Error::new(
- ErrorKind::Unexpected,
- "retryable_error from reader",
- )
- .set_temporary()),
- _ => Ok(RpDelete {}),
- },
- )
- })
- .collect(),
- ))),
- 4 => Err(
- Error::new(ErrorKind::Unexpected, "retryable_error from reader")
- .set_temporary(),
- ),
- 5 => Ok(RpBatch::new(BatchedResults::Delete(
- v.into_iter().map(|(s, _)| (s, Ok(RpDelete {}))).collect(),
- ))),
- _ => unreachable!(),
- },
+ _ => 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!(),
}
}
}
diff --git a/core/src/raw/rps.rs b/core/src/raw/rps.rs
index 05954afc..37fa37f3 100644
--- a/core/src/raw/rps.rs
+++ b/core/src/raw/rps.rs
@@ -132,71 +132,35 @@ impl RpRead {
/// Reply for `batch` operation.
pub struct RpBatch {
- results: BatchedResults,
+ results: Vec<(String, Result<BatchedReply>)>,
}
impl RpBatch {
/// Create a new RpBatch.
- pub fn new(results: BatchedResults) -> Self {
+ pub fn new(results: Vec<(String, Result<BatchedReply>)>) -> Self {
Self { results }
}
/// Get the results from RpBatch.
- pub fn results(&self) -> &BatchedResults {
+ pub fn results(&self) -> &[(String, Result<BatchedReply>)] {
&self.results
}
/// Consume RpBatch to get the batched results.
- pub fn into_results(self) -> BatchedResults {
+ pub fn into_results(self) -> Vec<(String, Result<BatchedReply>)> {
self.results
}
}
/// Batch results of `batch` operations.
-pub enum BatchedResults {
+pub enum BatchedReply {
/// results of delete batch operation
- Delete(Vec<(String, Result<RpDelete>)>),
+ Delete(RpDelete),
}
-impl BatchedResults {
- /// Return the length of given results.
- pub fn len(&self) -> usize {
- use BatchedResults::*;
- match self {
- Delete(v) => v.len(),
- }
- }
-
- /// Return if given results is empty.
- pub fn is_empty(&self) -> bool {
- use BatchedResults::*;
- match self {
- Delete(v) => v.is_empty(),
- }
- }
-
- /// Return the length of ok results.
- pub fn len_ok(&self) -> usize {
- use BatchedResults::*;
- match self {
- Delete(v) => v.iter().filter(|v| v.1.is_ok()).count(),
- }
- }
-
- /// Return the length of error results.
- pub fn len_err(&self) -> usize {
- use BatchedResults::*;
- match self {
- Delete(v) => v.iter().filter(|v| v.1.is_err()).count(),
- }
- }
-
- /// Return an iterator over error results.
- pub fn errors(&self) -> Box<dyn Iterator<Item = &Error> + '_> {
- use BatchedResults::*;
- match self {
- Delete(v) => Box::new(v.iter().filter_map(|v| v.1.as_ref().err())),
- }
+impl From<RpDelete> for BatchedReply {
+ fn from(rp: RpDelete) -> Self {
+ Self::Delete(rp)
}
}
diff --git a/core/src/services/azblob/backend.rs b/core/src/services/azblob/backend.rs
index c194be5d..78a421cb 100644
--- a/core/src/services/azblob/backend.rs
+++ b/core/src/services/azblob/backend.rs
@@ -598,59 +598,58 @@ impl Accessor for AzblobBackend {
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
- match ops {
- BatchOperations::Delete(ops) => {
- 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.azblob_batch_delete(&paths).await?;
+ 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.azblob_batch_delete(&paths).await?;
- // check response status
- if resp.status() != StatusCode::ACCEPTED {
- return Err(parse_error(resp).await?);
- }
+ // check response status
+ if resp.status() != StatusCode::ACCEPTED {
+ return Err(parse_error(resp).await?);
+ }
- // 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",
- )
- })?;
+ // 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 body = resp.into_body().bytes().await?;
- let body = String::from_utf8(body.to_vec()).map_err(|e| {
- Error::new(
- ErrorKind::Unexpected,
- &format!("get invalid batch response {e:?}"),
- )
- })?;
+ let body = resp.into_body().bytes().await?;
+ let body = String::from_utf8(body.to_vec()).map_err(|e| {
+ Error::new(
+ ErrorKind::Unexpected,
+ &format!("get invalid batch response {e:?}"),
+ )
+ })?;
- let results = parse_batch_delete_response(boundary, body, paths)?;
- Ok(RpBatch::new(BatchedResults::Delete(results)))
- }
- }
+ let results = parse_batch_delete_response(boundary, body, paths)?
+ .into_iter()
+ .map(|(path, rp)| (path, rp.map(|v| v.into())))
+ .collect();
+ Ok(RpBatch::new(results))
}
}
diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
index a61d7cdb..e8f59e75 100644
--- a/core/src/services/oss/backend.rs
+++ b/core/src/services/oss/backend.rs
@@ -544,61 +544,57 @@ impl Accessor for OssBackend {
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
- match ops {
- BatchOperations::Delete(ops) => {
- // Sadly, OSS will not return failed keys, so we will build
- // a set to calculate the failed keys.
- let mut keys = HashSet::new();
-
- let ops_len = ops.len();
- if ops_len > 1000 {
- return Err(Error::new(
- ErrorKind::Unsupported,
- "oss services only allow delete up to 1000 keys at once",
- )
- .with_context("length", ops_len.to_string()));
- }
-
- let paths = ops
- .into_iter()
- .map(|(p, _)| {
- keys.insert(p.clone());
- p
- })
- .collect();
-
- let resp = self.oss_delete_objects(paths).await?;
-
- let status = resp.status();
-
- if let StatusCode::OK = status {
- let bs = resp.into_body().bytes().await?;
-
- let result: DeleteObjectsResult = quick_xml::de::from_reader(bs.reader())
- .map_err(new_xml_deserialize_error)?;
-
- let mut batched_result = Vec::with_capacity(ops_len);
- for i in result.deleted {
- let path = build_rel_path(&self.root, &i.key);
- keys.remove(&path);
- batched_result.push((path, Ok(RpDelete::default())));
- }
- // TODO: we should handle those errors with code.
- for i in keys {
- batched_result.push((
- i,
- Err(Error::new(
- ErrorKind::Unexpected,
- "oss delete this key failed for reason we don't know",
- )),
- ));
- }
-
- Ok(RpBatch::new(BatchedResults::Delete(batched_result)))
- } else {
- Err(parse_error(resp).await?)
- }
+ // Sadly, OSS will not return failed keys, so we will build
+ // a set to calculate the failed keys.
+ let mut keys = HashSet::new();
+
+ let ops_len = ops.len();
+ if ops_len > 1000 {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "oss services only allow delete up to 1000 keys at once",
+ )
+ .with_context("length", ops_len.to_string()));
+ }
+
+ let paths = ops
+ .into_iter()
+ .map(|(p, _)| {
+ keys.insert(p.clone());
+ p
+ })
+ .collect();
+
+ let resp = self.oss_delete_objects(paths).await?;
+
+ let status = resp.status();
+
+ if let StatusCode::OK = status {
+ let bs = resp.into_body().bytes().await?;
+
+ let result: DeleteObjectsResult =
+ quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
+
+ let mut batched_result = Vec::with_capacity(ops_len);
+ for i in result.deleted {
+ let path = build_rel_path(&self.root, &i.key);
+ keys.remove(&path);
+ batched_result.push((path, Ok(RpDelete::default().into())));
}
+ // TODO: we should handle those errors with code.
+ for i in keys {
+ batched_result.push((
+ i,
+ Err(Error::new(
+ ErrorKind::Unexpected,
+ "oss delete this key failed for reason we don't know",
+ )),
+ ));
+ }
+
+ Ok(RpBatch::new(batched_result))
+ } else {
+ Err(parse_error(resp).await?)
}
}
}
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index 47098eb0..e1f9d7b8 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -1280,49 +1280,44 @@ impl Accessor for S3Backend {
async fn batch(&self, args: OpBatch) -> Result<RpBatch> {
let ops = args.into_operation();
- match ops {
- BatchOperations::Delete(ops) => {
- 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()));
- }
+ 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 paths = ops.into_iter().map(|(p, _)| p).collect();
- let resp = self.s3_delete_objects(paths).await?;
+ let resp = self.s3_delete_objects(paths).await?;
- let status = resp.status();
+ let status = resp.status();
- if let StatusCode::OK = status {
- let bs = resp.into_body().bytes().await?;
+ if let StatusCode::OK = status {
+ let bs = resp.into_body().bytes().await?;
- 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.root, &i.key);
- batched_result.push((path, Ok(RpDelete::default())));
- }
- // TODO: we should handle those errors with code.
- for i in result.error {
- let path = build_rel_path(&self.root, &i.key);
-
- batched_result.push((
- path,
- Err(Error::new(ErrorKind::Unexpected, &format!("{i:?}"))),
- ));
- }
-
- Ok(RpBatch::new(BatchedResults::Delete(batched_result)))
- } else {
- Err(parse_error(resp).await?)
- }
+ 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.root, &i.key);
+ batched_result.push((path, Ok(RpDelete::default().into())));
}
+ // TODO: we should handle those errors with code.
+ for i in result.error {
+ let path = build_rel_path(&self.root, &i.key);
+
+ batched_result.push((
+ path,
+ Err(Error::new(ErrorKind::Unexpected, &format!("{i:?}"))),
+ ));
+ }
+
+ Ok(RpBatch::new(batched_result))
+ } else {
+ Err(parse_error(resp).await?)
}
}
}
diff --git a/core/src/types/operator/operator.rs b/core/src/types/operator/operator.rs
index 0231dc88..791a0403 100644
--- a/core/src/types/operator/operator.rs
+++ b/core/src/types/operator/operator.rs
@@ -732,15 +732,16 @@ impl Operator {
/// ```
pub async fn remove_via(&self, input: impl Stream<Item = String> + Unpin) -> Result<()> {
if self.info().can_batch() {
- let mut input = input.map(|v| (v, OpDelete::default())).chunks(self.limit());
+ 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(BatchOperations::Delete(batches)))
- .await?;
-
- let BatchedResults::Delete(results) = results.into_results();
+ .batch(OpBatch::new(batches))
+ .await?
+ .into_results();
// TODO: return error here directly seems not a good idea?
for (_, result) in results {
@@ -805,15 +806,14 @@ impl Operator {
let batches = batches
.map_err(|err| err.1)?
.into_iter()
- .map(|v| (v.path().to_string(), OpDelete::default()))
+ .map(|v| (v.path().to_string(), OpDelete::default().into()))
.collect();
let results = self
.inner()
- .batch(OpBatch::new(BatchOperations::Delete(batches)))
- .await?;
-
- let BatchedResults::Delete(results) = results.into_results();
+ .batch(OpBatch::new(batches))
+ .await?
+ .into_results();
// TODO: return error here directly seems not a good idea?
for (_, result) in results {
diff --git a/core/src/types/ops.rs b/core/src/types/ops.rs
index 0612ac34..70addb8e 100644
--- a/core/src/types/ops.rs
+++ b/core/src/types/ops.rs
@@ -172,22 +172,22 @@ impl From<OpWrite> for PresignOperation {
/// Args for `batch` operation.
#[derive(Debug, Clone)]
pub struct OpBatch {
- ops: BatchOperations,
+ ops: Vec<(String, BatchOperation)>,
}
impl OpBatch {
/// Create a new batch options.
- pub fn new(ops: BatchOperations) -> Self {
+ pub fn new(ops: Vec<(String, BatchOperation)>) -> Self {
Self { ops }
}
/// Get operation from op.
- pub fn operation(&self) -> &BatchOperations {
+ pub fn operation(&self) -> &[(String, BatchOperation)] {
&self.ops
}
/// Consume OpBatch into BatchOperation
- pub fn into_operation(self) -> BatchOperations {
+ pub fn into_operation(self) -> Vec<(String, BatchOperation)> {
self.ops
}
}
@@ -195,35 +195,25 @@ impl OpBatch {
/// Batch operation used for batch.
#[derive(Debug, Clone)]
#[non_exhaustive]
-pub enum BatchOperations {
- /// Batch delete operations.
- Delete(Vec<(String, OpDelete)>),
+pub enum BatchOperation {
+ /// Batch delete operation.
+ Delete(OpDelete),
}
-impl BatchOperations {
+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 BatchOperations::*;
+ use BatchOperation::*;
match self {
Delete(_) => Operation::Delete,
}
}
-
- /// Return the length of given operations.
- pub fn len(&self) -> usize {
- use BatchOperations::*;
- match self {
- Delete(v) => v.len(),
- }
- }
-
- /// Return if given operations is empty.
- pub fn is_empty(&self) -> bool {
- use BatchOperations::*;
- match self {
- Delete(v) => v.is_empty(),
- }
- }
}
/// Args for `read` operation.