feat(storage): bump OpenDAL to v0.51

Cargo.toml

@@ -157,7 +157,7 @@ icelake = { git = "https://github.com/risingwavelabs/icelake.git", rev = "0ec44f
 iceberg = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "683fb89edeaf8d1baae69e1f376d68b92be1d496", features = ["storage-s3", "storage-gcs"] }
 iceberg-catalog-rest = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "683fb89edeaf8d1baae69e1f376d68b92be1d496" }
 iceberg-catalog-glue = { git = "https://github.com/risingwavelabs/iceberg-rust.git", rev = "683fb89edeaf8d1baae69e1f376d68b92be1d496" }
-opendal = "0.49"
+opendal = "0.51"
 # used only by arrow-udf-flight
 arrow-flight = "53"
 arrow-udf-js = "0.5"

src/connector/src/source/filesystem/opendal_source/opendal_enumerator.rs

@@ -19,7 +19,7 @@ use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use futures::stream::{self, BoxStream};
 use futures::StreamExt;
-use opendal::{Metakey, Operator};
+use opendal::Operator;
 use risingwave_common::types::Timestamptz;
 
 use super::OpendalSource;
@@ -69,12 +69,7 @@ impl<Src: OpendalSource> OpendalEnumerator<Src> {
     pub async fn list(&self) -> ConnectorResult<ObjectMetadataIter> {
         let prefix = self.prefix.as_deref().unwrap_or("/");
 
-        let object_lister = self
-            .op
-            .lister_with(prefix)
-            .recursive(true)
-            .metakey(Metakey::ContentLength | Metakey::LastModified)
-            .await?;
+        let object_lister = self.op.lister_with(prefix).recursive(true).await?;
         let stream = stream::unfold(object_lister, |mut object_lister| async move {
             match object_lister.next().await {
                 Some(Ok(object)) => {

src/object_store/src/object/opendal_engine/opendal_object_store.rs

@@ -22,7 +22,7 @@ use futures::{stream, StreamExt};
 use opendal::layers::{RetryLayer, TimeoutLayer};
 use opendal::raw::BoxedStaticFuture;
 use opendal::services::Memory;
-use opendal::{Execute, Executor, Metakey, Operator, Writer};
+use opendal::{Execute, Executor, Operator, Writer};
 use risingwave_common::config::ObjectStoreConfig;
 use risingwave_common::range::RangeBoundsExt;
 use thiserror_ext::AsReport;
@@ -227,7 +227,7 @@ impl ObjectStore for OpendalObjectStore {
     /// Deletes the objects with the given paths permanently from the storage. If an object
     /// specified in the request is not found, it will be considered as successfully deleted.
     async fn delete_objects(&self, paths: &[String]) -> ObjectResult<()> {
-        self.op.remove(paths.to_vec()).await?;
+        self.op.delete_iter(paths.to_vec()).await?;
         Ok(())
     }
 
@@ -237,11 +237,7 @@ impl ObjectStore for OpendalObjectStore {
         start_after: Option<String>,
         limit: Option<usize>,
     ) -> ObjectResult<ObjectMetadataIter> {
-        let mut object_lister = self
-            .op
-            .lister_with(prefix)
-            .recursive(true)
-            .metakey(Metakey::ContentLength);
+        let mut object_lister = self.op.lister_with(prefix).recursive(true);
         if let Some(start_after) = start_after {
             object_lister = object_lister.start_after(&start_after);
         }

src/object_store/src/object/opendal_engine/opendal_s3.rs

@@ -108,7 +108,7 @@ impl OpendalObjectStore {
         if let Some(nodelay) = config.s3.nodelay.as_ref() {
             client_builder = client_builder.tcp_nodelay(*nodelay);
         }
-
+        #[allow(deprecated)]
         Ok(HttpClient::build(client_builder)?)
     }