Merge pull request #7629 from Xuanwo/bump-opendal

feat: Bump OpenDAL to v0.17

Author: Xuanwo

Cargo.lock

@@ -12,4 +12,4 @@ test = false
 common-base = { path = "../base" }
 
 async-trait = "0.1.56"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }

src/common/contexts/Cargo.toml

@@ -83,18 +83,14 @@ impl Accessor for DalContext {
             .metadata()
     }
 
-    fn presign(&self, args: &OpPresign) -> Result<PresignedRequest> {
-        self.get_inner()?.presign(args)
+    async fn create(&self, path: &str, args: OpCreate) -> Result<()> {
+        self.get_inner()?.create(path, args).await
     }
 
-    async fn create(&self, args: &OpCreate) -> Result<()> {
-        self.get_inner()?.create(args).await
-    }
-
-    async fn read(&self, args: &OpRead) -> Result<BytesReader> {
+    async fn read(&self, path: &str, args: OpRead) -> Result<BytesReader> {
         let metric = self.metrics.clone();
 
-        self.get_inner()?.read(args).await.map(|r| {
+        self.get_inner()?.read(path, args).await.map(|r| {
             let mut last_pending = None;
             let r = observe_read(r, move |e| {
                 let start = match last_pending {
@@ -117,7 +113,7 @@ impl Accessor for DalContext {
         })
     }
 
-    async fn write(&self, args: &OpWrite, r: BytesReader) -> Result<u64> {
+    async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
         let metric = self.metrics.clone();
 
         let mut last_pending = None;
@@ -139,18 +135,22 @@ impl Accessor for DalContext {
             metric.inc_write_bytes_cost(start.elapsed().as_millis() as u64);
         });
 
-        self.get_inner()?.write(args, Box::new(r)).await
+        self.get_inner()?.write(path, args, Box::new(r)).await
+    }
+
+    async fn stat(&self, path: &str, args: OpStat) -> Result<ObjectMetadata> {
+        self.get_inner()?.stat(path, args).await
     }
 
-    async fn stat(&self, args: &OpStat) -> Result<ObjectMetadata> {
-        self.get_inner()?.stat(args).await
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<()> {
+        self.get_inner()?.delete(path, args).await
     }
 
-    async fn delete(&self, args: &OpDelete) -> Result<()> {
-        self.get_inner()?.delete(args).await
+    async fn list(&self, path: &str, args: OpList) -> Result<DirStreamer> {
+        self.get_inner()?.list(path, args).await
     }
 
-    async fn list(&self, args: &OpList) -> Result<DirStreamer> {
-        self.get_inner()?.list(args).await
+    fn presign(&self, path: &str, args: OpPresign) -> Result<PresignedRequest> {
+        self.get_inner()?.presign(path, args)
     }
 }

src/common/contexts/src/dal/dal_context.rs

@@ -83,61 +83,61 @@ impl Accessor for DalRuntime {
             .metadata()
     }
 
-    fn presign(&self, args: &OpPresign) -> Result<PresignedRequest> {
-        self.get_inner()?.presign(args)
-    }
-
-    async fn create(&self, args: &OpCreate) -> Result<()> {
+    async fn create(&self, path: &str, args: OpCreate) -> Result<()> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.create(&args).await })
+            .spawn(async move { op.create(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn read(&self, args: &OpRead) -> Result<BytesReader> {
+    async fn read(&self, path: &str, args: OpRead) -> Result<BytesReader> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.read(&args).await })
+            .spawn(async move { op.read(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn write(&self, args: &OpWrite, r: BytesReader) -> Result<u64> {
+    async fn write(&self, path: &str, args: OpWrite, r: BytesReader) -> Result<u64> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.write(&args, r).await })
+            .spawn(async move { op.write(&path, args, r).await })
             .await
             .expect("join must success")
     }
 
-    async fn stat(&self, args: &OpStat) -> Result<ObjectMetadata> {
+    async fn stat(&self, path: &str, args: OpStat) -> Result<ObjectMetadata> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.stat(&args).await })
+            .spawn(async move { op.stat(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn delete(&self, args: &OpDelete) -> Result<()> {
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<()> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.delete(&args).await })
+            .spawn(async move { op.delete(&path, args).await })
             .await
             .expect("join must success")
     }
 
-    async fn list(&self, args: &OpList) -> Result<DirStreamer> {
+    async fn list(&self, path: &str, args: OpList) -> Result<DirStreamer> {
         let op = self.get_inner()?;
-        let args = args.clone();
+        let path = path.to_string();
         self.runtime
-            .spawn(async move { op.list(&args).await })
+            .spawn(async move { op.list(&path, args).await })
             .await
             .expect("join must success")
     }
+
+    fn presign(&self, path: &str, args: OpPresign) -> Result<PresignedRequest> {
+        self.get_inner()?.presign(path, args)
+    }
 }

src/common/contexts/src/dal/dal_runtime.rs

@@ -18,7 +18,7 @@ anyhow = "1.0.64"
 backon = "0.1.0"
 globiter = "0.1.0"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = [
+opendal = { version = "0.17.0", features = [
     "layers-retry",
     "layers-tracing",
     "layers-metrics",

src/common/storage/Cargo.toml

@@ -26,5 +26,5 @@ common-users = { path = "../users" }
 
 async-trait = "0.1.56"
 dyn-clone = "1.0.6"
-opendal = { version = "0.16.0", features = ["layers-retry"] }
+opendal = { version = "0.17.0", features = ["layers-retry"] }
 parking_lot = "0.12.1"

src/query/catalog/Cargo.toml

@@ -21,7 +21,7 @@ common-users = { path = "../users" }
 clap = { version = "3.2.5", features = ["derive", "env"] }
 hex = "0.4.3"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = ["layers-retry", "compress"], optional = true }
+opendal = { version = "0.17.0", features = ["layers-retry", "compress"], optional = true }
 semver = "1.0.10"
 serde = { version = "1.0.137", features = ["derive"] }
 serfig = "0.0.2"

src/query/config/Cargo.toml

@@ -23,5 +23,5 @@ common-streams = { path = "../../streams" }
 async-trait = { version = "0.1.0", package = "async-trait-fn" }
 futures = "0.3.21"
 futures-util = "0.3.21"
-opendal = { version = "0.16.0", features = ["layers-retry", "compress"] }
+opendal = { version = "0.17.0", features = ["layers-retry", "compress"] }
 parking_lot = "0.12.1"

src/query/pipeline/sources/Cargo.toml

@@ -111,7 +111,7 @@ num = "0.4.0"
 num_cpus = "1.13.1"
 octocrab = "0.16.0"
 once_cell = "1.12.0"
-opendal = { version = "0.16.0", features = ["layers-retry", "layers-tracing", "layers-metrics", "compress"] }
+opendal = { version = "0.17.0", features = ["layers-retry", "layers-tracing", "layers-metrics", "compress"] }
 opensrv-mysql = "0.2.0"
 openssl = { version = "0.10.40", features = ["vendored"] }
 parking_lot = "0.12.1"

src/query/service/Cargo.toml

@@ -227,14 +227,14 @@ where
     P: Debug + Send,
     P: Iterator<Item = IOError>,
 {
-    async fn read(&self, _args: &OpRead) -> std::io::Result<BytesReader> {
+    async fn read(&self, _: &str, _: OpRead) -> std::io::Result<BytesReader> {
         let v = &mut (*self.err.lock());
         let err = v.take_err();
         v.increase_read_op_count();
         Err(err)
     }
 
-    async fn write(&self, _args: &OpWrite, _r: BytesReader) -> std::io::Result<u64> {
+    async fn write(&self, _: &str, _: OpWrite, _: BytesReader) -> std::io::Result<u64> {
         let v = &mut (*self.err.lock());
         let err = v.take_err();
         v.increase_write_op_count();