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 f7f9990f9 fix(integration/object_store): object_store requires
metadata in list (#5501)
f7f9990f9 is described below
commit f7f9990f95146400c3aef8afc11804e4a6e3afe3
Author: Xuanwo <[email protected]>
AuthorDate: Fri Jan 3 15:49:35 2025 +0800
fix(integration/object_store): object_store requires metadata in list
(#5501)
---
integrations/object_store/src/store.rs | 73 +++++++++++++++++++++++-----------
integrations/object_store/src/utils.rs | 12 +-----
2 files changed, 51 insertions(+), 34 deletions(-)
diff --git a/integrations/object_store/src/store.rs
b/integrations/object_store/src/store.rs
index c5ba9a675..32ccbc2d8 100644
--- a/integrations/object_store/src/store.rs
+++ b/integrations/object_store/src/store.rs
@@ -323,32 +323,52 @@ impl ObjectStore for OpendalStore {
let offset = offset.clone();
let fut = async move {
- let fut = if
self.inner.info().full_capability().list_with_start_after {
- self.inner
- .lister_with(&path)
- .start_after(offset.as_ref())
- .recursive(true)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, &path))?
- .then(try_format_object_meta)
- .into_send()
- .boxed()
+ let list_with_start_after =
self.inner.info().full_capability().list_with_start_after;
+ let mut fut = self.inner.lister_with(&path).recursive(true);
+
+ // Use native start_after support if possible.
+ if list_with_start_after {
+ fut = fut.start_after(offset.as_ref());
+ }
+
+ let lister = fut
+ .await
+ .map_err(|err| format_object_store_error(err, &path))?
+ .then(move |entry| {
+ let path = path.clone();
+
+ async move {
+ let entry = entry.map_err(|err|
format_object_store_error(err, &path))?;
+ let (path, metadata) = entry.into_parts();
+
+ // If it's a dir or last_modified is present, we can
use it directly.
+ if metadata.is_dir() ||
metadata.last_modified().is_some() {
+ let object_meta = format_object_meta(&path,
&metadata);
+ return Ok(object_meta);
+ }
+
+ let metadata = self
+ .inner
+ .stat(&path)
+ .await
+ .map_err(|err| format_object_store_error(err,
&path))?;
+ let object_meta = format_object_meta(&path, &metadata);
+ Ok::<_, object_store::Error>(object_meta)
+ }
+ })
+ .into_send()
+ .boxed();
+
+ let stream = if list_with_start_after {
+ lister
} else {
- self.inner
- .lister_with(&path)
- .recursive(true)
- .into_future()
- .into_send()
- .await
- .map_err(|err| format_object_store_error(err, &path))?
- .try_filter(move |entry|
futures::future::ready(entry.path() > offset.as_ref()))
- .then(try_format_object_meta)
+ lister
+ .try_filter(move |entry|
futures::future::ready(entry.location > offset))
.into_send()
.boxed()
};
- Ok::<_, object_store::Error>(fut)
+
+ Ok::<_, object_store::Error>(stream)
};
fut.into_stream().into_send().try_flatten().boxed()
@@ -374,8 +394,15 @@ impl ObjectStore for OpendalStore {
if meta.is_dir() {
common_prefixes.push(entry.path().into());
- } else {
+ } else if meta.last_modified().is_some() {
objects.push(format_object_meta(entry.path(), meta));
+ } else {
+ let meta = self
+ .inner
+ .stat(entry.path())
+ .await
+ .map_err(|err| format_object_store_error(err,
entry.path()))?;
+ objects.push(format_object_meta(entry.path(), &meta));
}
}
diff --git a/integrations/object_store/src/utils.rs
b/integrations/object_store/src/utils.rs
index 5eb4a10f7..63357402b 100644
--- a/integrations/object_store/src/utils.rs
+++ b/integrations/object_store/src/utils.rs
@@ -17,7 +17,7 @@
use futures::Stream;
use object_store::ObjectMeta;
-use opendal::{Entry, Metadata};
+use opendal::Metadata;
use std::future::IntoFuture;
/// Conditionally add the `Send` marker trait for the wrapped type.
@@ -60,16 +60,6 @@ pub fn format_object_meta(path: &str, meta: &Metadata) ->
ObjectMeta {
}
}
-/// Try to format `opendal::Entry` to `object_store::ObjectMeta`.
-pub async fn try_format_object_meta(
- res: object_store::Result<Entry, opendal::Error>,
-) -> object_store::Result<ObjectMeta> {
- let entry = res.map_err(|err| format_object_store_error(err, ""))?;
- let meta = entry.metadata();
-
- Ok(format_object_meta(entry.path(), meta))
-}
-
/// Make given future `Send`.
pub trait IntoSendFuture {
type Output;
