refactor: Bump OpenDAL to 0.48.0 (#16147)

* refactor: Bump OpenDAL to 0.48.0

Signed-off-by: Xuanwo <github@xuanwo.io>

* format toml

Signed-off-by: Xuanwo <github@xuanwo.io>

---------

Signed-off-by: Xuanwo <github@xuanwo.io>
Co-authored-by: Bohu <overred.shuttler@gmail.com>

Cargo.toml

@@ -255,10 +255,11 @@ itertools = "0.10.5"
 jsonb = "0.4.1"
 match-template = "0.0.1"
 mysql_async = { version = "0.34", default-features = false, features = ["rustls-tls"] }
+object_store_opendal = "0.45"
 once_cell = "1.15.0"
 openai_api_rust = "0.1"
-opendal = { version = "0.47.1", features = [
-    "layers-minitrace",
+opendal = { version = "0.48.0", features = [
+    "layers-fastrace",
     "layers-prometheus-client",
     "layers-async-backtrace",
     "services-s3",
@@ -322,10 +323,10 @@ walkdir = "2.3.2"
 xorfilter-rs = "0.5"
 
 # Observability
+fastrace = { version = "0.6", features = ["enable"] }
+fastrace-opentelemetry = "0.6"
 log = { version = "0.4.21", features = ["serde", "kv_unstable_std"] }
 logcall = "0.1.9"
-minitrace = { version = "0.6.6", features = ["enable"] }
-minitrace-opentelemetry = "0.6.6"
 opentelemetry = { version = "0.23", features = ["trace", "logs"] }
 opentelemetry-otlp = { version = "0.16", features = [
     "trace",
@@ -399,7 +400,6 @@ backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "7226
 color-eyre = { git = "https://github.com/eyre-rs/eyre.git", rev = "e5d92c3" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "57795da" }
 ethnum = { git = "https://github.com/ariesdevil/ethnum-rs", rev = "4cb05f1" }
-object_store_opendal = { package = "object_store_opendal", git = "https://github.com/Xuanwo/opendal", rev = "d6baf68" }
 openai_api_rust = { git = "https://github.com/datafuse-extras/openai-api", rev = "819a0ed" }
 orc-rust = { git = "https://github.com/youngsofun/datafusion-orc", rev = "1745375" }
 recursive = { git = "https://github.com/zhang2014/recursive.git", rev = "6af35a1" }

src/binaries/tool/table_meta_inspector.rs

@@ -91,8 +91,7 @@ async fn parse_input_data(config: &InspectorConfig) -> Result<Vec<u8>> {
                 }
                 None => {
                     let current_dir = env::current_dir()?;
-                    let mut builder = Fs::default();
-                    builder.root(current_dir.to_str().ok_or("Invalid path")?);
+                    let builder = Fs::default().root(current_dir.to_str().ok_or("Invalid path")?);
                     Operator::new(builder)?.finish()
                 }
             };

src/common/base/Cargo.toml

@@ -27,11 +27,11 @@ bytesize = "1.1.0"
 ctrlc = { version = "3.2.3", features = ["termination"] }
 databend-common-exception = { workspace = true }
 enquote = "1.1.0"
+fastrace = { workspace = true }
 futures = { workspace = true }
 libc = "0.2.153"
 log = { workspace = true }
 logcall = { workspace = true }
-minitrace = { workspace = true }
 num_cpus = "1.13.1"
 once_cell = { workspace = true }
 parking_lot = { workspace = true }

src/common/base/src/containers/pool.rs

@@ -119,7 +119,7 @@ where
     /// When returning an existent one, `check()` will be called on it to ensure it is still valid.
     /// E.g., when returning a tcp connection.
     #[logcall::logcall(err = "debug")]
-    #[minitrace::trace]
+    #[fastrace::trace]
     pub async fn get(&self, key: &Mgr::Key) -> Result<Mgr::Item, Mgr::Error> {
         let pool_item = self.get_pool_item(key);
 

src/common/storage/src/metrics.rs

@@ -240,10 +240,11 @@ impl<R: oio::BlockingRead> oio::BlockingRead for StorageMetricsWrapper<R> {
 }
 
 impl<R: oio::Write> oio::Write for StorageMetricsWrapper<R> {
-    async fn write(&mut self, bs: Buffer) -> Result<usize> {
+    async fn write(&mut self, bs: Buffer) -> Result<()> {
         let start = Instant::now();
+        let size = bs.len();
 
-        self.inner.write(bs).await.inspect(|&size| {
+        self.inner.write(bs).await.inspect(|_| {
             self.metrics.inc_write_bytes(size);
             self.metrics
                 .inc_write_bytes_cost(start.elapsed().as_millis() as u64);
@@ -260,10 +261,11 @@ impl<R: oio::Write> oio::Write for StorageMetricsWrapper<R> {
 }
 
 impl<R: oio::BlockingWrite> oio::BlockingWrite for StorageMetricsWrapper<R> {
-    fn write(&mut self, bs: Buffer) -> Result<usize> {
+    fn write(&mut self, bs: Buffer) -> Result<()> {
         let start = Instant::now();
+        let size = bs.len();
 
-        self.inner.write(bs).inspect(|&size| {
+        self.inner.write(bs).inspect(|_| {
             self.metrics.inc_write_bytes(size);
             self.metrics
                 .inc_write_bytes_cost(start.elapsed().as_millis() as u64);

src/common/storage/src/metrics_layer.rs

@@ -516,14 +516,15 @@ impl<R: oio::BlockingRead> oio::BlockingRead for OperatorMetricsWrapper<R> {
 }
 
 impl<R: oio::Write> oio::Write for OperatorMetricsWrapper<R> {
-    async fn write(&mut self, bs: Buffer) -> opendal::Result<usize> {
+    async fn write(&mut self, bs: Buffer) -> opendal::Result<()> {
         let start = Instant::now();
+        let size = bs.len();
 
         self.inner
             .write(bs)
             .await
-            .inspect(|&res| {
-                self.metrics.observe_bytes_total(self.scheme, self.op, res);
+            .inspect(|_| {
+                self.metrics.observe_bytes_total(self.scheme, self.op, size);
                 self.metrics
                     .observe_request_duration(self.scheme, self.op, start.elapsed());
             })
@@ -549,13 +550,14 @@ impl<R: oio::Write> oio::Write for OperatorMetricsWrapper<R> {
 }
 
 impl<R: oio::BlockingWrite> oio::BlockingWrite for OperatorMetricsWrapper<R> {
-    fn write(&mut self, bs: Buffer) -> opendal::Result<usize> {
+    fn write(&mut self, bs: Buffer) -> opendal::Result<()> {
         let start = Instant::now();
+        let size = bs.len();
 
         self.inner
             .write(bs)
-            .inspect(|&res| {
-                self.metrics.observe_bytes_total(self.scheme, self.op, res);
+            .inspect(|_| {
+                self.metrics.observe_bytes_total(self.scheme, self.op, size);
                 self.metrics
                     .observe_request_duration(self.scheme, self.op, start.elapsed());
             })