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 2026/05/21 10:09:52 UTC
(opendal) branch main updated: feat(services/tos): add versioning support (#7575)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 05b3c8e02 feat(services/tos): add versioning support (#7575)
05b3c8e02 is described below
commit 05b3c8e02f5d187239cb618bfdffbd16830f5f5d
Author: Xin Sun <dd...@gmail.com>
AuthorDate: Thu May 21 18:09:44 2026 +0800
feat(services/tos): add versioning support (#7575)
Change-Id: I99c7c230a797a249894fc7656a57ba9cb827f9da
---
core/services/tos/src/backend.rs | 22 +++++--
core/services/tos/src/core.rs | 103 +++++++++++++++++++++++++++++++-
core/services/tos/src/deleter.rs | 37 ++++++++++--
core/services/tos/src/lister.rs | 126 +++++++++++++++++++++++++++++++++++++++
4 files changed, 276 insertions(+), 12 deletions(-)
diff --git a/core/services/tos/src/backend.rs b/core/services/tos/src/backend.rs
index 01a93ba39..6e6aaec2e 100644
--- a/core/services/tos/src/backend.rs
+++ b/core/services/tos/src/backend.rs
@@ -24,7 +24,7 @@ use crate::core::constants::{X_TOS_DIRECTORY, X_TOS_META_PREFIX};
use crate::core::*;
use crate::deleter::TosDeleter;
use crate::error::parse_error;
-use crate::lister::TosLister;
+use crate::lister::{TosLister, TosListers, TosObjectVersionsLister};
use crate::utils::tos_parse_into_metadata;
use crate::writer::TosWriter;
use http::Response;
@@ -203,6 +203,8 @@ impl Builder for TosBuilder {
list_with_limit: true,
list_with_start_after: true,
list_with_recursive: true,
+ list_with_versions: true,
+ list_with_deleted: true,
stat: true,
stat_with_if_match: true,
@@ -252,7 +254,7 @@ pub struct TosBackend {
impl Access for TosBackend {
type Reader = HttpBody;
type Writer = oio::MultipartWriter<TosWriter>;
- type Lister = oio::PageLister<TosLister>;
+ type Lister = TosListers;
type Deleter = oio::BatchDeleter<TosDeleter>;
type Copier = ();
@@ -336,8 +338,20 @@ impl Access for TosBackend {
}
async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let lister = TosLister::new(self.core.clone(), path, args);
- Ok((RpList::default(), oio::PageLister::new(lister)))
+ let lister = if args.versions() || args.deleted() {
+ TwoWays::Two(oio::PageLister::new(TosObjectVersionsLister::new(
+ self.core.clone(),
+ path,
+ args,
+ )))
+ } else {
+ TwoWays::One(oio::PageLister::new(TosLister::new(
+ self.core.clone(),
+ path,
+ args,
+ )))
+ };
+ Ok((RpList::default(), lister))
}
async fn copy(
diff --git a/core/services/tos/src/core.rs b/core/services/tos/src/core.rs
index 3477552fb..445d64a84 100644
--- a/core/services/tos/src/core.rs
+++ b/core/services/tos/src/core.rs
@@ -413,6 +413,7 @@ impl TosCore {
version_id: op.version().map(|v| v.to_owned()),
})
.collect(),
+ quiet: false,
})
.map_err(new_json_serialize_error)?;
@@ -506,6 +507,47 @@ impl TosCore {
self.send(req).await
}
+ pub async fn tos_list_object_versions(
+ &self,
+ prefix: &str,
+ delimiter: &str,
+ limit: Option<usize>,
+ key_marker: &str,
+ version_id_marker: &str,
+ ) -> Result<Response<Buffer>> {
+ let p = build_abs_path(&self.root, prefix);
+
+ let mut url =
+ QueryPairsWriter::new(&format!("https://{}.{}", self.bucket, self.endpoint_domain));
+ url = url.push("versions", "");
+
+ if !p.is_empty() {
+ url = url.push("prefix", &percent_encode_query(&p));
+ }
+ if !delimiter.is_empty() {
+ url = url.push("delimiter", &percent_encode_query(delimiter));
+ }
+ if let Some(limit) = limit {
+ url = url.push("max-keys", &limit.to_string());
+ }
+ if !key_marker.is_empty() {
+ url = url.push("key-marker", &percent_encode_query(key_marker));
+ }
+ if !version_id_marker.is_empty() {
+ url = url.push(
+ "version-id-marker",
+ &percent_encode_query(version_id_marker),
+ );
+ }
+
+ let req = Request::get(url.finish())
+ .extension(Operation::List)
+ .body(Buffer::new())
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
pub async fn tos_initiate_multipart_upload(
&self,
path: &str,
@@ -685,13 +727,14 @@ pub struct CompleteMultipartUploadResult {
}
#[derive(Default, Debug, Serialize)]
-#[serde(default, rename_all = "camelCase")]
+#[serde(default, rename_all = "PascalCase")]
pub struct DeleteObjectsRequest {
pub objects: Vec<DeleteObjectsRequestObject>,
+ pub quiet: bool,
}
#[derive(Default, Debug, Serialize)]
-#[serde(rename_all = "camelCase")]
+#[serde(rename_all = "PascalCase")]
pub struct DeleteObjectsRequestObject {
pub key: String,
#[serde(skip_serializing_if = "Option::is_none")]
@@ -701,15 +744,29 @@ pub struct DeleteObjectsRequestObject {
#[derive(Default, Debug, Deserialize)]
#[serde(default, rename_all = "PascalCase")]
pub struct DeleteObjectsResult {
+ pub deleted: Vec<DeleteObjectsResultDeleted>,
+ #[serde(alias = "Error")]
pub errors: Vec<DeleteObjectsResultError>,
}
+#[derive(Default, Debug, Deserialize)]
+#[serde(default, rename_all = "PascalCase")]
+pub struct DeleteObjectsResultDeleted {
+ pub key: String,
+ #[serde(alias = "VersionID")]
+ pub version_id: Option<String>,
+ pub delete_marker: bool,
+ #[serde(alias = "DeleteMarkerVersionID")]
+ pub delete_marker_version_id: Option<String>,
+}
+
#[derive(Default, Debug, Deserialize)]
#[serde(default, rename_all = "PascalCase")]
pub struct DeleteObjectsResultError {
pub code: String,
pub key: String,
pub message: String,
+ #[serde(alias = "VersionID")]
pub version_id: Option<String>,
}
@@ -742,3 +799,45 @@ pub struct ListObjectsOutputContent {
pub struct OutputCommonPrefix {
pub prefix: String,
}
+
+#[derive(Default, Debug, Deserialize)]
+#[serde(default, rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutput {
+ pub name: String,
+ pub prefix: String,
+ pub key_marker: String,
+ #[serde(alias = "VersionIDMarker")]
+ pub version_id_marker: String,
+ pub max_keys: usize,
+ pub delimiter: String,
+ pub is_truncated: Option<bool>,
+ pub next_key_marker: Option<String>,
+ #[serde(alias = "NextVersionIDMarker")]
+ pub next_version_id_marker: Option<String>,
+ pub common_prefixes: Vec<OutputCommonPrefix>,
+ pub versions: Vec<ListObjectVersionsOutputVersion>,
+ pub delete_markers: Vec<ListObjectVersionsOutputDeleteMarker>,
+}
+
+#[derive(Default, Debug, Deserialize)]
+#[serde(default, rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutputVersion {
+ pub key: String,
+ #[serde(alias = "VersionID")]
+ pub version_id: String,
+ pub is_latest: bool,
+ pub last_modified: String,
+ #[serde(rename = "ETag")]
+ pub etag: Option<String>,
+ pub size: u64,
+}
+
+#[derive(Default, Debug, Deserialize)]
+#[serde(default, rename_all = "PascalCase")]
+pub struct ListObjectVersionsOutputDeleteMarker {
+ pub key: String,
+ #[serde(alias = "VersionID")]
+ pub version_id: String,
+ pub is_latest: bool,
+ pub last_modified: String,
+}
diff --git a/core/services/tos/src/deleter.rs b/core/services/tos/src/deleter.rs
index 4a43ec517..4811c076a 100644
--- a/core/services/tos/src/deleter.rs
+++ b/core/services/tos/src/deleter.rs
@@ -24,6 +24,7 @@ use opendal_core::raw::*;
use opendal_core::*;
use crate::core::*;
+use crate::error::parse_tos_error_code;
pub struct TosDeleter {
core: Arc<TosCore>,
@@ -65,9 +66,10 @@ impl oio::BatchDelete for TosDeleter {
let result: DeleteObjectsResult =
serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
+ let mut deleted = result.deleted;
let mut errors = result.errors;
let mut batched_result = BatchDeleteResult {
- succeeded: Vec::with_capacity(batch.len() - errors.len()),
+ succeeded: Vec::with_capacity(batch.len().saturating_sub(errors.len())),
failed: Vec::with_capacity(errors.len()),
};
for (path, op) in batch {
@@ -77,11 +79,24 @@ impl oio::BatchDelete for TosDeleter {
.position(|e| e.key == abs_path && e.version_id.as_deref() == op.version())
{
let error = errors.swap_remove(idx);
- batched_result.failed.push((
- path,
- op,
- Error::new(ErrorKind::Unexpected, error.message),
- ));
+ batched_result
+ .failed
+ .push((path, op, parse_delete_objects_result_error(error)));
+ } else if let Some(idx) = deleted.iter().position(|e| {
+ e.key == abs_path
+ && (op.version().is_none() || e.version_id.as_deref() == op.version())
+ }) {
+ let deleted = deleted.swap_remove(idx);
+ let op = if op.version().is_some() {
+ if let Some(version) = &deleted.version_id {
+ op.with_version(version)
+ } else {
+ op
+ }
+ } else {
+ op
+ };
+ batched_result.succeeded.push((path, op));
} else {
batched_result.succeeded.push((path, op));
}
@@ -90,3 +105,13 @@ impl oio::BatchDelete for TosDeleter {
Ok(batched_result)
}
}
+
+fn parse_delete_objects_result_error(err: DeleteObjectsResultError) -> Error {
+ let (kind, retryable) =
+ parse_tos_error_code(err.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
+ let mut err = Error::new(kind, format!("{err:?}"));
+ if retryable {
+ err = err.set_temporary();
+ }
+ err
+}
diff --git a/core/services/tos/src/lister.rs b/core/services/tos/src/lister.rs
index 4f2071950..9439e921b 100644
--- a/core/services/tos/src/lister.rs
+++ b/core/services/tos/src/lister.rs
@@ -26,6 +26,8 @@ use opendal_core::Metadata;
use opendal_core::Result;
use opendal_core::raw::*;
+pub type TosListers = TwoWays<oio::PageLister<TosLister>, oio::PageLister<TosObjectVersionsLister>>;
+
pub struct TosLister {
core: Arc<TosCore>,
@@ -112,3 +114,127 @@ impl oio::PageList for TosLister {
Ok(())
}
}
+
+pub struct TosObjectVersionsLister {
+ core: Arc<TosCore>,
+
+ prefix: String,
+ args: OpList,
+
+ delimiter: &'static str,
+ abs_start_after: Option<String>,
+}
+
+impl TosObjectVersionsLister {
+ pub fn new(core: Arc<TosCore>, path: &str, args: OpList) -> Self {
+ let delimiter = if args.recursive() { "" } else { "/" };
+ let abs_start_after = args
+ .start_after()
+ .map(|start_after| build_abs_path(&core.root, start_after));
+
+ Self {
+ core,
+ prefix: path.to_string(),
+ args,
+ delimiter,
+ abs_start_after,
+ }
+ }
+}
+
+impl oio::PageList for TosObjectVersionsLister {
+ async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
+ let markers = ctx.token.rsplit_once(" ");
+ let (key_marker, version_id_marker) = if let Some(data) = markers {
+ data
+ } else if let Some(start_after) = &self.abs_start_after {
+ (start_after.as_str(), "")
+ } else {
+ ("", "")
+ };
+
+ let resp = self
+ .core
+ .tos_list_object_versions(
+ &self.prefix,
+ self.delimiter,
+ self.args.limit(),
+ key_marker,
+ version_id_marker,
+ )
+ .await?;
+
+ if resp.status() != http::StatusCode::OK {
+ return Err(parse_error(resp));
+ }
+
+ let bs = resp.into_body();
+ let output: ListObjectVersionsOutput =
+ serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
+
+ ctx.done = if let Some(is_truncated) = output.is_truncated {
+ !is_truncated
+ } else {
+ false
+ };
+ ctx.token = format!(
+ "{} {}",
+ output.next_key_marker.unwrap_or_default(),
+ output.next_version_id_marker.unwrap_or_default()
+ );
+
+ for prefix in output.common_prefixes {
+ let de = oio::Entry::new(
+ &build_rel_path(&self.core.root, &prefix.prefix),
+ Metadata::new(EntryMode::DIR),
+ );
+
+ ctx.entries.push_back(de);
+ }
+
+ for version_object in output.versions {
+ if !(self.args.versions() || version_object.is_latest) {
+ continue;
+ }
+
+ let mut path = build_rel_path(&self.core.root, &version_object.key);
+ if path.is_empty() {
+ path = "/".to_string();
+ }
+
+ let mut meta = Metadata::new(EntryMode::from_path(&path));
+ meta.set_version(&version_object.version_id);
+ meta.set_is_current(version_object.is_latest);
+ meta.set_content_length(version_object.size);
+ meta.set_last_modified(version_object.last_modified.parse::<Timestamp>()?);
+
+ if let Some(etag) = version_object.etag {
+ meta.set_etag(&etag);
+ meta.set_content_md5(etag.trim_matches('"'));
+ }
+
+ let de = oio::Entry::with(path, meta);
+ ctx.entries.push_back(de);
+ }
+
+ if self.args.deleted() {
+ for delete_marker in output.delete_markers {
+ let mut path = build_rel_path(&self.core.root, &delete_marker.key);
+ if path.is_empty() {
+ path = "/".to_string();
+ }
+
+ let mut meta = Metadata::new(EntryMode::FILE);
+ meta.set_version(&delete_marker.version_id);
+ meta.set_is_deleted(true);
+ meta.set_is_current(delete_marker.is_latest);
+ meta.set_last_modified(delete_marker.last_modified.parse::<Timestamp>()?);
+
+ let de = oio::Entry::with(path, meta);
+ ctx.entries.push_back(de);
+ }
+ }
+
+ Ok(())
+ }
+}