From 04d2184f76e670c6f5dcb26e624ea08823e989a3 Mon Sep 17 00:00:00 2001
From: Xuanwo <github@xuanwo.io>
Date: Fri, 3 Jan 2025 15:35:24 +0800
Subject: [PATCH 1/2] fix(integration/object_store): object_store requires
metadata in list
Signed-off-by: Xuanwo <github@xuanwo.io>
---
integrations/object_store/src/store.rs | 64 +++++++++++++++++---------
integrations/object_store/src/utils.rs | 12 +----
2 files changed, 43 insertions(+), 33 deletions(-)
diff --git a/integrations/object_store/src/store.rs b/integrations/object_store/src/store.rs
index c5ba9a6750f7..8b64dd35f03d 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()
diff --git a/integrations/object_store/src/utils.rs b/integrations/object_store/src/utils.rs
index 5eb4a10f73bc..63357402b7a0 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;
From c1ad35547ee5325ea2f285415c9c988f97e1b9a4 Mon Sep 17 00:00:00 2001
From: Xuanwo <github@xuanwo.io>
Date: Fri, 3 Jan 2025 15:38:21 +0800
Subject: [PATCH 2/2] Fix
Signed-off-by: Xuanwo <github@xuanwo.io>
---
integrations/object_store/src/store.rs | 9 ++++++++-
1 file changed, 8 insertions(+), 1 deletion(-)
diff --git a/integrations/object_store/src/store.rs b/integrations/object_store/src/store.rs
index 8b64dd35f03d..32ccbc2d8a23 100644
--- a/integrations/object_store/src/store.rs
+++ b/integrations/object_store/src/store.rs
@@ -394,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));
}
}