chore(query): revert "fix(query): update opendal (#19110)" (#19146)

* revert

* revert

* revert

Author: zhang2014

Cargo.lock

@@ -312,13 +312,13 @@ hyper-util = { version = "0.1.9", features = ["client", "client-legacy", "tokio"
 lru = "0.12"
 
 ## in branch dev
-iceberg = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "6536f9c", features = [
+iceberg = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403", features = [
     "storage-all",
 ] }
-iceberg-catalog-glue = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "6536f9c" }
-iceberg-catalog-hms = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "6536f9c" }
-iceberg-catalog-rest = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "6536f9c" }
-iceberg-catalog-s3tables = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "6536f9c" }
+iceberg-catalog-glue = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
+iceberg-catalog-hms = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
+iceberg-catalog-rest = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
+iceberg-catalog-s3tables = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "32b1403" }
 
 # Explicitly specify compatible AWS SDK versions
 aws-config = "1.5.18"
@@ -367,9 +367,9 @@ num-derive = "0.4.2"
 num-traits = "0.2.19"
 num_cpus = "1.17"
 object = "0.36.5"
-object_store_opendal = { git = "https://github.com/apache/opendal.git", rev = "02953ef" }
+object_store_opendal = { version = "0.54.1" }
 once_cell = "1.15.0"
-opendal = { git = "https://github.com/apache/opendal.git", rev = "02953ef", features = [
+opendal = { version = "0.54.1", features = [
     "layers-fastrace",
     "layers-prometheus-client",
     "layers-async-backtrace",
@@ -387,8 +387,6 @@ opendal = { git = "https://github.com/apache/opendal.git", rev = "02953ef", feat
     "services-webhdfs",
     "services-huggingface",
 ] }
-opendal-layer-immutable-index = { git = "https://github.com/apache/opendal.git", rev = "02953ef" }
-opendal-layer-observe-metrics-common = { git = "https://github.com/apache/opendal.git", rev = "02953ef" }
 openraft = { version = "0.10.0", features = [
     "serde",
     "tracing-log",

Cargo.toml

@@ -219,7 +219,7 @@ mod tests {
     use std::path::Path;
 
     use databend_common_base::base::tokio;
-    use databend_common_storage::Scheme;
+    use opendal::Scheme;
 
     use super::*;
 
@@ -237,12 +237,12 @@ mod tests {
     #[tokio::test]
     async fn test_load_epochfs_storage() -> Result<()> {
         let op = load_bendsave_storage("s3://bendsave/tmp?region=us-east-1").await?;
-        assert_eq!(op.info().scheme(), Scheme::S3.to_string());
+        assert_eq!(op.info().scheme(), Scheme::S3);
         assert_eq!(op.info().name(), "bendsave");
         assert_eq!(op.info().root(), "/tmp/");
 
         let op = load_bendsave_storage("fs://opt").await?;
-        assert_eq!(op.info().scheme(), Scheme::Fs.to_string());
+        assert_eq!(op.info().scheme(), Scheme::Fs);
         assert_eq!(op.info().root(), "/opt");
         Ok(())
     }

src/bendsave/src/storage.rs

@@ -183,19 +183,12 @@ impl From<std::convert::Infallible> for ErrorCode {
 
 impl From<opendal::Error> for ErrorCode {
     fn from(error: opendal::Error) -> Self {
-        let msg = error.message();
-        let detail = error.to_string();
-        let detail = detail
-            .strip_suffix(msg)
-            .and_then(|err| err.strip_suffix(" => "))
-            .unwrap_or(&detail);
-
         match error.kind() {
-            opendal::ErrorKind::NotFound => ErrorCode::StorageNotFound(msg).add_detail(detail),
+            opendal::ErrorKind::NotFound => ErrorCode::StorageNotFound(error.to_string()),
             opendal::ErrorKind::PermissionDenied => {
-                ErrorCode::StoragePermissionDenied(msg).add_detail(detail)
+                ErrorCode::StoragePermissionDenied(error.to_string())
             }
-            _ => ErrorCode::StorageOther(msg).add_detail(detail),
+            _ => ErrorCode::StorageOther(format!("{error:?}")),
         }
     }
 }

src/common/exception/src/exception_into.rs

@@ -34,8 +34,6 @@ iceberg = { workspace = true }
 log = { workspace = true }
 lru = { workspace = true }
 opendal = { workspace = true }
-opendal-layer-immutable-index = { workspace = true }
-opendal-layer-observe-metrics-common = { workspace = true }
 parquet = { workspace = true }
 prometheus-client = { workspace = true }
 regex = { workspace = true }

src/common/storage/Cargo.toml

@@ -43,7 +43,6 @@ pub use http_client::StorageHttpClient;
 mod operator;
 pub use operator::DataOperator;
 pub use operator::OperatorRegistry;
-pub use operator::Scheme;
 pub use operator::check_operator;
 pub use operator::init_operator;
 

src/common/storage/src/lib.rs

@@ -23,9 +23,9 @@ use databend_common_base::runtime::metrics::FamilyHistogram;
 use databend_common_base::runtime::metrics::register_counter_family;
 use databend_common_base::runtime::metrics::register_gauge_family;
 use databend_common_base::runtime::metrics::register_histogram_family;
+use opendal::layers::observe;
 use opendal::raw::Access;
 use opendal::raw::Layer;
-use opendal_layer_observe_metrics_common as observe;
 use prometheus_client::encoding::EncodeLabel;
 use prometheus_client::encoding::EncodeLabelSet;
 use prometheus_client::encoding::LabelSetEncoder;

src/common/storage/src/metrics_layer.rs

@@ -16,7 +16,6 @@ use std::env;
 use std::io::Error;
 use std::io::ErrorKind;
 use std::io::Result;
-use std::str::FromStr;
 use std::sync::LazyLock;
 use std::time::Duration;
 
@@ -53,13 +52,13 @@ use opendal::layers::AsyncBacktraceLayer;
 use opendal::layers::ConcurrentLimitLayer;
 use opendal::layers::FastraceLayer;
 use opendal::layers::HttpClientLayer;
+use opendal::layers::ImmutableIndexLayer;
 use opendal::layers::LoggingLayer;
 use opendal::layers::RetryInterceptor;
 use opendal::layers::RetryLayer;
 use opendal::layers::TimeoutLayer;
 use opendal::raw::HttpClient;
 use opendal::services;
-use opendal_layer_immutable_index::ImmutableIndexLayer;
 
 use crate::StorageConfig;
 use crate::StorageHttpClient;
@@ -404,8 +403,8 @@ fn init_s3_operator(cfg: &StorageS3Config) -> Result<impl Builder> {
         .session_token(&cfg.security_token)
         .role_arn(&cfg.role_arn)
         .external_id(&cfg.external_id)
-        // Don't enable it otherwise we will get Permission in stat unknown files
-        // .allow_anonymous()
+        // It's safe to allow anonymous since opendal will perform the check first.
+        .allow_anonymous()
         // Root.
         .root(&cfg.root);
 
@@ -666,76 +665,3 @@ impl OperatorRegistry for iceberg::io::FileIO {
         Ok((file_io.get_operator().clone(), &location[pos..]))
     }
 }
-
-#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
-pub enum Scheme {
-    Azblob,
-    Gcs,
-    Hdfs,
-    Ipfs,
-    S3,
-    Oss,
-    Obs,
-    Cos,
-    Http,
-    Fs,
-    Webhdfs,
-    Huggingface,
-    Custom(&'static str),
-}
-
-impl Scheme {
-    /// Convert self into static str.
-    pub fn into_static(self) -> &'static str {
-        self.into()
-    }
-}
-
-impl From<Scheme> for &'static str {
-    fn from(v: Scheme) -> Self {
-        match v {
-            Scheme::Azblob => "azblob",
-            Scheme::Gcs => "gcs",
-            Scheme::Hdfs => "hdfs",
-            Scheme::Ipfs => "ipfs",
-            Scheme::S3 => "s3",
-            Scheme::Oss => "oss",
-            Scheme::Obs => "obs",
-            Scheme::Cos => "cos",
-            Scheme::Http => "http",
-            Scheme::Fs => "fs",
-            Scheme::Webhdfs => "webhdfs",
-            Scheme::Huggingface => "huggingface",
-            Scheme::Custom(s) => s,
-        }
-    }
-}
-
-impl FromStr for Scheme {
-    type Err = Error;
-
-    fn from_str(s: &str) -> Result<Self> {
-        let s = s.to_lowercase();
-        match s.as_str() {
-            "azblob" => Ok(Scheme::Azblob),
-            "gcs" => Ok(Scheme::Gcs),
-            "hdfs" => Ok(Scheme::Hdfs),
-            "ipfs" => Ok(Scheme::Ipfs),
-            "s3" | "s3a" => Ok(Scheme::S3),
-            "oss" => Ok(Scheme::Oss),
-            "obs" => Ok(Scheme::Obs),
-            "cos" => Ok(Scheme::Cos),
-            "http" | "https" => Ok(Scheme::Http),
-            "fs" => Ok(Scheme::Fs),
-            "webhdfs" => Ok(Scheme::Webhdfs),
-            "huggingface" | "hf" => Ok(Scheme::Huggingface),
-            _ => Ok(Scheme::Custom(Box::leak(s.into_boxed_str()))),
-        }
-    }
-}
-
-impl std::fmt::Display for Scheme {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.into_static())
-    }
-}

src/common/storage/src/operator.rs

@@ -307,18 +307,26 @@ impl<R: oio::List> oio::List for RuntimeIO<R> {
 }
 
 impl<R: oio::Delete> oio::Delete for RuntimeIO<R> {
-    async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
-        self.inner.as_mut().unwrap().delete(path, args).await
+    fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> {
+        self.inner.as_mut().unwrap().delete(path, args)
     }
 
-    async fn close(&mut self) -> Result<()> {
+    async fn flush(&mut self) -> Result<usize> {
         let mut r = self.inner.take().expect("deleter must be valid");
         let runtime = self.runtime.clone();
 
-        let _ = runtime
-            .spawn(async move { r.close().await })
+        let (r, res) = runtime
+            .try_spawn(
+                async move {
+                    let res = r.flush().await;
+                    (r, res)
+                },
+                Some(self.spawn_task_name.clone()),
+            )
+            .expect("spawn must success")
             .await
-            .expect("join must success")?;
-        Ok(())
+            .expect("join must success");
+        self.inner = Some(r);
+        res
     }
 }

src/common/storage/src/runtime_layer.rs

@@ -60,10 +60,7 @@ impl StageFileInfo {
             path,
             size: meta.content_length(),
             md5: meta.content_md5().map(str::to_string),
-            last_modified: meta.last_modified().map(|m| {
-                let ns = m.into_inner().as_nanosecond();
-                DateTime::from_timestamp_nanos(ns as i64)
-            }),
+            last_modified: meta.last_modified(),
             etag: meta.etag().map(str::to_string),
             status: StageFileStatus::NeedCopy,
             creator: None,