feat(core): add version(bool) for List operation to include version d… (

#5106)

* feat(core): add version(bool) for List operation to include version during list or not

* fix docs

* fix cargo clippy

* rename bucket_versioning_enabled to enable_versioning && support start_after

* enable bucket versioning in ceph_rados_s3 action

* add versioning test action for minio and ceph_radios

* disable ceph radios with versioning action

* handle markers internally in S3

.github/services/s3/0_minio_s3/action.yml

@@ -31,7 +31,9 @@ runs:
  AWS_ACCESS_KEY_ID: "minioadmin"
  AWS_SECRET_ACCESS_KEY: "minioadmin"
  AWS_EC2_METADATA_DISABLED: "true"
- run: aws --endpoint-url http://127.0.0.1:9000/ s3 mb s3://test
+ run: |
+ aws --endpoint-url http://127.0.0.1:9000/ s3 mb s3://test
+
  - name: Setup
  shell: bash
  run: |

.github/services/s3/ceph_radios_s3_with_versioning/disable_action.yml

@@ -0,0 +1,47 @@
+# 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.
+
+name: ceph_rados_s3_with_versioning
+description: 'Behavior test for CEPH OBJECT GATEWAY S3 with bucket versioning enabled'
+
+runs:
+ using: "composite"
+ steps:
+ - name: Setup Ceph Rados Server
+ shell: bash
+ working-directory: fixtures/s3
+ run: docker compose -f docker-compose-ceph-rados.yml up -d --wait
+
+ # ceph/demo has support for CEPH_DEMO_BUCKET, but it doesn't work as expected.
+ - name: Create bucket
+ shell: bash
+ working-directory: fixtures/s3
+ run: |
+ docker exec ceph-demo s3cmd mb s3://demo
+ docker exec ceph-demo s3cmd setversioning s3://demo enable
+
+ - name: Setup
+ shell: bash
+ run: |
+ cat << EOF >> $GITHUB_ENV
+ OPENDAL_S3_BUCKET=demo
+ OPENDAL_S3_ENDPOINT=http://127.0.0.1:8080
+ OPENDAL_S3_ACCESS_KEY_ID=demo
+ OPENDAL_S3_SECRET_ACCESS_KEY=demo
+ OPENDAL_S3_REGION=us-east-1
+ OPENDAL_S3_ENABLE_VERSIONING=true
+ EOF

.github/services/s3/minio_s3_with_versioning/action.yml

@@ -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.
+
+name: minio_s3_with_versioning
+description: 'Behavior test for Minio S3 with bucket versioning enabled'
+
+runs:
+ using: "composite"
+ steps:
+ - name: Setup MinIO Server
+ shell: bash
+ working-directory: fixtures/s3
+ run: docker compose -f docker-compose-minio.yml up -d --wait
+ - name: Setup test bucket
+ shell: bash
+ env:
+ AWS_ACCESS_KEY_ID: "minioadmin"
+ AWS_SECRET_ACCESS_KEY: "minioadmin"
+ AWS_EC2_METADATA_DISABLED: "true"
+ run: |
+ aws --endpoint-url http://127.0.0.1:9000/ s3 mb s3://test
+ aws --endpoint-url http://127.0.0.1:9000/ s3api put-bucket-versioning --bucket test --versioning-configuration Status=Enabled
+
+ - name: Setup
+ shell: bash
+ run: |
+ cat << EOF >> $GITHUB_ENV
+ OPENDAL_S3_BUCKET=test
+ OPENDAL_S3_ENDPOINT=http://127.0.0.1:9000
+ OPENDAL_S3_ACCESS_KEY_ID=minioadmin
+ OPENDAL_S3_SECRET_ACCESS_KEY=minioadmin
+ OPENDAL_S3_REGION=us-east-1
+ OPENDAL_S3_ENABLE_VERSIONING=true
+ EOF

core/src/raw/enum_utils.rs

@@ -92,6 +92,15 @@ impl<ONE: oio::Write, TWO: oio::Write> oio::Write for TwoWays<ONE, TWO> {
  }
 }
 
+impl<ONE: oio::List, TWO: oio::List> oio::List for TwoWays<ONE, TWO> {
+ async fn next(&mut self) -> Result<Option<oio::Entry>> {
+ match self {
+ Self::One(v) => v.next().await,
+ Self::Two(v) => v.next().await,
+ }
+ }
+}
+
 /// ThreeWays is used to implement traits that based on three ways.
 ///
 /// Users can wrap three different trait types together.

core/src/raw/oio/list/page_list.rs

@@ -15,11 +15,10 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::collections::VecDeque;
-use std::future::Future;
-
 use crate::raw::*;
 use crate::*;
+use std::collections::VecDeque;
+use std::future::Future;
 
 /// PageList is used to implement [`oio::List`] based on API supporting pagination. By implementing
 /// PageList, services don't need to care about the details of page list.

core/src/raw/ops.rs

@@ -22,10 +22,9 @@
 use std::collections::HashMap;
 use std::time::Duration;
 
-use flagset::FlagSet;
-
 use crate::raw::*;
 use crate::*;
+use flagset::FlagSet;
 
 /// Args for `create` operation.
 ///
@@ -100,6 +99,14 @@ pub struct OpList {
  /// - If this is set to > 1, the list operation will be concurrent,
  /// and the maximum number of concurrent operations will be determined by this value.
  concurrent: usize,
+ /// The version is used to control whether the object versions should be returned.
+ ///
+ /// - If `false`, list operation will not return with object versions
+ /// - If `true`, list operation will return with object versions if object versioning is supported
+ /// by the underlying service
+ ///
+ /// Default to `false`
+ version: bool,
 }
 
 impl Default for OpList {
@@ -111,6 +118,7 @@ impl Default for OpList {
  // By default, we want to know what's the mode of this entry.
  metakey: Metakey::Mode.into(),
  concurrent: 1,
+ version: false,
  }
  }
 }
@@ -184,6 +192,17 @@ impl OpList {
  pub fn concurrent(&self) -> usize {
  self.concurrent
  }
+
+ /// Change the version of this list operation
+ pub fn with_version(mut self, version: bool) -> Self {
+ self.version = version;
+ self
+ }
+
+ /// Get the version of this list operation
+ pub fn version(&self) -> bool {
+ self.version
+ }
 }
 
 /// Args for `presign` operation.
@@ -333,36 +352,36 @@ impl OpRead {
  &mut self.range
  }
 
- /// Sets the content-disposition header that should be send back by the remote read operation.
+ /// Sets the content-disposition header that should be sent back by the remote read operation.
  pub fn with_override_content_disposition(mut self, content_disposition: &str) -> Self {
  self.override_content_disposition = Some(content_disposition.into());
  self
  }
 
- /// Returns the content-disposition header that should be send back by the remote read
+ /// Returns the content-disposition header that should be sent back by the remote read
  /// operation.
  pub fn override_content_disposition(&self) -> Option<&str> {
  self.override_content_disposition.as_deref()
  }
 
- /// Sets the cache-control header that should be send back by the remote read operation.
+ /// Sets the cache-control header that should be sent back by the remote read operation.
  pub fn with_override_cache_control(mut self, cache_control: &str) -> Self {
  self.override_cache_control = Some(cache_control.into());
  self
  }
 
- /// Returns the cache-control header that should be send back by the remote read operation.
+ /// Returns the cache-control header that should be sent back by the remote read operation.
  pub fn override_cache_control(&self) -> Option<&str> {
  self.override_cache_control.as_deref()
  }
 
- /// Sets the content-type header that should be send back by the remote read operation.
+ /// Sets the content-type header that should be sent back by the remote read operation.
  pub fn with_override_content_type(mut self, content_type: &str) -> Self {
  self.override_content_type = Some(content_type.into());
  self
  }
 
- /// Returns the content-type header that should be send back by the remote read operation.
+ /// Returns the content-type header that should be sent back by the remote read operation.
  pub fn override_content_type(&self) -> Option<&str> {
  self.override_content_type.as_deref()
  }
@@ -526,13 +545,13 @@ impl OpStat {
  self.if_none_match.as_deref()
  }
 
- /// Sets the content-disposition header that should be send back by the remote read operation.
+ /// Sets the content-disposition header that should be sent back by the remote read operation.
  pub fn with_override_content_disposition(mut self, content_disposition: &str) -> Self {
  self.override_content_disposition = Some(content_disposition.into());
  self
  }
 
- /// Returns the content-disposition header that should be send back by the remote read
+ /// Returns the content-disposition header that should be sent back by the remote read
  /// operation.
  pub fn override_content_disposition(&self) -> Option<&str> {
  self.override_content_disposition.as_deref()

core/src/services/s3/backend.rs

@@ -43,9 +43,10 @@ use reqwest::Url;
 use super::core::*;
 use super::error::parse_error;
 use super::error::parse_s3_error_code;
-use super::lister::S3Lister;
+use super::lister::{S3Lister, S3Listers, S3ObjectVersionsLister};
 use super::writer::S3Writer;
 use super::writer::S3Writers;
+use crate::raw::oio::PageLister;
 use crate::raw::*;
 use crate::services::S3Config;
 use crate::*;
@@ -457,6 +458,13 @@ impl S3Builder {
  self
  }
 
+ /// Set bucket versioning status for this backend
+ pub fn enable_versioning(mut self, enabled: bool) -> Self {
+ self.config.enable_versioning = enabled;
+
+ self
+ }
+
  /// Check if `bucket` is valid
  /// `bucket` must be not empty and if `enable_virtual_host_style` is true
  /// it couldn't contain dot(.) character
@@ -856,6 +864,7 @@ impl Builder for S3Builder {
  default_storage_class,
  allow_anonymous: self.config.allow_anonymous,
  disable_stat_with_override: self.config.disable_stat_with_override,
+ enable_versioning: self.config.enable_versioning,
  signer,
  loader,
  credential_loaded: AtomicBool::new(false),
@@ -876,7 +885,7 @@ pub struct S3Backend {
 impl Access for S3Backend {
  type Reader = HttpBody;
  type Writer = S3Writers;
- type Lister = oio::PageLister<S3Lister>;
+ type Lister = S3Listers;
  type BlockingReader = ();
  type BlockingWriter = ();
  type BlockingLister = ();
@@ -929,6 +938,7 @@ impl Access for S3Backend {
  list_with_limit: true,
  list_with_start_after: true,
  list_with_recursive: true,
+ list_with_version: self.core.enable_versioning,
 
  presign: true,
  presign_stat: true,
@@ -1027,14 +1037,32 @@ impl Access for S3Backend {
  }
 
  async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
- let l = S3Lister::new(
- self.core.clone(),
- path,
- args.recursive(),
- args.limit(),
- args.start_after(),
- );
- Ok((RpList::default(), oio::PageLister::new(l)))
+ if args.version() && !self.core.enable_versioning {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "the bucket doesn't enable versioning",
+ ));
+ }
+
+ let l = if args.version() {
+ TwoWays::Two(PageLister::new(S3ObjectVersionsLister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ )))
+ } else {
+ TwoWays::One(PageLister::new(S3Lister::new(
+ self.core.clone(),
+ path,
+ args.recursive(),
+ args.limit(),
+ args.start_after(),
+ )))
+ };
+
+ Ok((RpList::default(), l))
  }
 
  async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {

core/src/services/s3/config.rs

@@ -36,6 +36,8 @@ pub struct S3Config {
  ///
  /// required.
  pub bucket: String,
+ /// is bucket versioning enabled for this bucket
+ pub enable_versioning: bool,
  /// endpoint of this backend.
  ///
  /// Endpoint must be full uri, e.g.