chore: bump opendal to 0.54.1 (#18970)

* chore: remove blocking IO

* chore: bump iceberg-rs's opendal

* chore: remove synchronous read methods and update opendal dependency

* chore: clean

* fix: unit test

* fix: tests

* chore: replace url

* chore: replace url

* fix: ensure max_files does not exceed available files

Author: dqhl76

Cargo.lock

@@ -330,13 +330,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 = "d5cca1c15f240f3cb04e57569bce648933b1c79b", features = [
+iceberg = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "1dace26ea25a9b9e2066367cbd3b7badc75dd7f9", features = [
     "storage-all",
 ] }
-iceberg-catalog-glue = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "d5cca1c15f240f3cb04e57569bce648933b1c79b" }
-iceberg-catalog-hms = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "d5cca1c15f240f3cb04e57569bce648933b1c79b" }
-iceberg-catalog-rest = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "d5cca1c15f240f3cb04e57569bce648933b1c79b" }
-iceberg-catalog-s3tables = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "d5cca1c15f240f3cb04e57569bce648933b1c79b" }
+iceberg-catalog-glue = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "1dace26ea25a9b9e2066367cbd3b7badc75dd7f9" }
+iceberg-catalog-hms = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "1dace26ea25a9b9e2066367cbd3b7badc75dd7f9" }
+iceberg-catalog-rest = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "1dace26ea25a9b9e2066367cbd3b7badc75dd7f9" }
+iceberg-catalog-s3tables = { version = "0.4.0", git = "https://github.com/databendlabs/iceberg-rust", rev = "1dace26ea25a9b9e2066367cbd3b7badc75dd7f9" }
 
 # Explicitly specify compatible AWS SDK versions
 aws-config = "1.5.18"
@@ -385,13 +385,12 @@ num-derive = "0.4.2"
 num-traits = "0.2.19"
 num_cpus = "1.17"
 object = "0.36.5"
-object_store_opendal = { version = "0.52.0" }
+object_store_opendal = { version = "0.54.1" }
 once_cell = "1.15.0"
-opendal = { version = "0.53.2", features = [
+opendal = { version = "0.54.1", features = [
     "layers-fastrace",
     "layers-prometheus-client",
     "layers-async-backtrace",
-    "layers-blocking",
     "services-s3",
     "services-fs",
     "services-gcs",

Cargo.toml

@@ -164,13 +164,9 @@ pub struct StorageMetricsAccessor<A: Access> {
 impl<A: Access> LayeredAccess for StorageMetricsAccessor<A> {
     type Inner = A;
     type Reader = StorageMetricsWrapper<A::Reader>;
-    type BlockingReader = StorageMetricsWrapper<A::BlockingReader>;
     type Writer = StorageMetricsWrapper<A::Writer>;
-    type BlockingWriter = StorageMetricsWrapper<A::BlockingWriter>;
     type Lister = A::Lister;
-    type BlockingLister = A::BlockingLister;
     type Deleter = A::Deleter;
-    type BlockingDeleter = A::BlockingDeleter;
 
     fn inner(&self) -> &Self::Inner {
         &self.inner
@@ -201,26 +197,6 @@ impl<A: Access> LayeredAccess for StorageMetricsAccessor<A> {
     async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
         self.inner.delete().await
     }
-
-    fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
-        self.inner
-            .blocking_read(path, args)
-            .map(|(rp, r)| (rp, StorageMetricsWrapper::new(r, self.metrics.clone())))
-    }
-
-    fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
-        self.inner
-            .blocking_write(path, args)
-            .map(|(rp, r)| (rp, StorageMetricsWrapper::new(r, self.metrics.clone())))
-    }
-
-    fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
-        self.inner.blocking_list(path, args)
-    }
-
-    fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
-        self.inner.blocking_delete()
-    }
 }
 
 pub struct StorageMetricsWrapper<R> {
@@ -246,12 +222,6 @@ impl<R: oio::Read> oio::Read for StorageMetricsWrapper<R> {
     }
 }
 
-impl<R: oio::BlockingRead> oio::BlockingRead for StorageMetricsWrapper<R> {
-    fn read(&mut self) -> Result<Buffer> {
-        self.inner.read()
-    }
-}
-
 impl<R: oio::Write> oio::Write for StorageMetricsWrapper<R> {
     async fn write(&mut self, bs: Buffer) -> Result<()> {
         let start = Instant::now();
@@ -272,20 +242,3 @@ impl<R: oio::Write> oio::Write for StorageMetricsWrapper<R> {
         self.inner.abort().await
     }
 }
-
-impl<R: oio::BlockingWrite> oio::BlockingWrite for StorageMetricsWrapper<R> {
-    fn write(&mut self, bs: Buffer) -> Result<()> {
-        let start = Instant::now();
-        let size = bs.len();
-
-        self.inner.write(bs).inspect(|_| {
-            self.metrics.inc_write_bytes(size);
-            self.metrics
-                .inc_write_bytes_cost(start.elapsed().as_millis() as u64);
-        })
-    }
-
-    fn close(&mut self) -> Result<Metadata> {
-        self.inner.close()
-    }
-}

src/common/storage/src/metrics.rs

@@ -263,7 +263,7 @@ impl std::hash::Hash for OperationLabels {
 
 impl EncodeLabelSet for OperationLabels {
     fn encode(&self, mut encoder: LabelSetEncoder) -> Result<(), fmt::Error> {
-        (observe::LABEL_SCHEME, self.0.scheme.into_static()).encode(encoder.encode_label())?;
+        (observe::LABEL_SCHEME, self.0.scheme).encode(encoder.encode_label())?;
         (observe::LABEL_NAMESPACE, self.0.namespace.as_ref()).encode(encoder.encode_label())?;
         (observe::LABEL_OPERATION, self.0.operation).encode(encoder.encode_label())?;
 

src/common/storage/src/metrics_layer.rs

@@ -49,6 +49,7 @@ use log::warn;
 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;
@@ -201,7 +202,7 @@ fn build_operator<B: Builder>(builder: B, cfg: Option<&StorageNetworkParams>) ->
         .finish();
 
     // Make sure the http client has been updated.
-    ob.update_http_client(|_| HttpClient::with(get_http_client(cfg)));
+    let ob = ob.layer(HttpClientLayer::new(HttpClient::with(get_http_client(cfg))));
 
     let mut op = ob
         // Add retry

src/common/storage/src/operator.rs

@@ -42,15 +42,6 @@ pub async fn read_parquet_schema_async_rs(
     infer_schema_with_extension(meta.file_metadata())
 }
 
-pub fn read_parquet_schema_sync_rs(
-    operator: &Operator,
-    path: &str,
-    file_size: Option<u64>,
-) -> Result<ArrowSchema> {
-    let meta = read_metadata_sync(path, operator, file_size)?;
-    infer_schema_with_extension(meta.file_metadata())
-}
-
 pub fn infer_schema_with_extension(meta: &FileMetaData) -> Result<ArrowSchema> {
     let mut arrow_schema = parquet_to_arrow_schema(meta.schema_descr(), meta.key_value_metadata())?;
     // Convert data types to extension types using meta information.
@@ -146,54 +137,6 @@ pub async fn read_metadata_async(
     }
 }
 
-pub fn read_metadata_sync(
-    path: &str,
-    operator: &Operator,
-    file_size: Option<u64>,
-) -> Result<ParquetMetaData> {
-    let blocking = operator.blocking();
-    let file_size = match file_size {
-        None => blocking.stat(path)?.content_length(),
-        Some(n) => n,
-    };
-
-    check_footer_size(file_size, path)?;
-
-    let map_err =
-        |e: ParquetError| ErrorCode::BadBytes(format!("Invalid Parquet file '{path}': {e}",));
-    // read and cache up to DEFAULT_FOOTER_READ_SIZE bytes from the end and process the footer
-    let default_end_len = DEFAULT_FOOTER_READ_SIZE.min(file_size);
-    let buffer = blocking
-        .read_with(path)
-        .range((file_size - default_end_len)..file_size)
-        .call()?
-        .to_vec();
-    let buffer_len = buffer.len();
-    let footer_tail = ParquetMetaDataReader::decode_footer_tail(
-        &buffer[(buffer_len - FOOTER_SIZE as usize)..]
-            .try_into()
-            .unwrap(),
-    )
-    .map_err(map_err)?;
-    let metadata_len = footer_tail.metadata_length() as u64;
-    check_meta_size(file_size, metadata_len, path)?;
-
-    let footer_len = FOOTER_SIZE + metadata_len;
-    if (footer_len as usize) <= buffer_len {
-        // The whole metadata is in the bytes we already read
-        let offset = buffer_len - footer_len as usize;
-        Ok(ParquetMetaDataReader::decode_metadata(&buffer[offset..]).map_err(map_err)?)
-    } else {
-        let mut metadata = blocking
-            .read_with(path)
-            .range((file_size - footer_len)..(file_size - buffer_len as u64))
-            .call()?
-            .to_vec();
-        metadata.extend(buffer);
-        Ok(ParquetMetaDataReader::decode_metadata(&metadata).map_err(map_err)?)
-    }
-}
-
 /// check file is large enough to hold footer
 fn check_footer_size(file_size: u64, path: &str) -> Result<()> {
     if file_size < FOOTER_SIZE {

src/common/storage/src/parquet.rs

@@ -91,13 +91,9 @@ impl<A> Debug for RuntimeAccessor<A> {
 impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
     type Inner = A;
     type Reader = RuntimeIO<A::Reader>;
-    type BlockingReader = A::BlockingReader;
     type Writer = RuntimeIO<A::Writer>;
-    type BlockingWriter = A::BlockingWriter;
     type Lister = RuntimeIO<A::Lister>;
-    type BlockingLister = A::BlockingLister;
     type Deleter = RuntimeIO<A::Deleter>;
-    type BlockingDeleter = A::BlockingDeleter;
 
     fn inner(&self) -> &Self::Inner {
         &self.inner
@@ -182,22 +178,6 @@ impl<A: Access> LayeredAccess for RuntimeAccessor<A> {
             .await
             .expect("join must success")
     }
-
-    fn blocking_read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::BlockingReader)> {
-        self.inner.blocking_read(path, args)
-    }
-
-    fn blocking_write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::BlockingWriter)> {
-        self.inner.blocking_write(path, args)
-    }
-
-    fn blocking_list(&self, path: &str, args: OpList) -> Result<(RpList, Self::BlockingLister)> {
-        self.inner.blocking_list(path, args)
-    }
-
-    fn blocking_delete(&self) -> Result<(RpDelete, Self::BlockingDeleter)> {
-        self.inner.blocking_delete()
-    }
 }
 
 pub struct RuntimeIO<R: 'static> {

src/common/storage/src/runtime_layer.rs

@@ -130,7 +130,7 @@ impl StageFilesInfo {
         mut files: &[String],
     ) -> Result<Vec<StageFileInfo>> {
         if let Some(m) = max_files {
-            files = &files[..m]
+            files = &files[..m.min(files.len())]
         }
         let file_infos = self.stat_concurrent(operator, thread_num, files).await?;
         let mut res = Vec::with_capacity(file_infos.len());
@@ -207,39 +207,6 @@ impl StageFilesInfo {
         Ok(files.pop())
     }
 
-    pub fn blocking_list(
-        &self,
-        operator: &Operator,
-        max_files: Option<usize>,
-    ) -> Result<Vec<StageFileInfo>> {
-        let max_files = max_files.unwrap_or(usize::MAX);
-        if let Some(files) = &self.files {
-            let mut res = Vec::new();
-            for file in files {
-                let full_path = Path::new(&self.path)
-                    .join(file)
-                    .to_string_lossy()
-                    .trim_start_matches('/')
-                    .to_string();
-                let meta = operator.blocking().stat(&full_path)?;
-                if meta.mode().is_file() {
-                    res.push(StageFileInfo::new(full_path, &meta))
-                } else {
-                    return Err(ErrorCode::BadArguments(format!(
-                        "{full_path} is not a file"
-                    )));
-                }
-                if res.len() == max_files {
-                    return Ok(res);
-                }
-            }
-            Ok(res)
-        } else {
-            let pattern = self.get_pattern()?;
-            blocking_list_files_with_pattern(operator, &self.path, pattern, max_files)
-        }
-    }
-
     #[async_backtrace::framed]
     pub async fn list_files_with_pattern(
         operator: &Operator,
@@ -374,52 +341,6 @@ fn check_file(path: &str, mode: EntryMode, pattern: &Option<Regex>) -> bool {
     }
 }
 
-fn blocking_list_files_with_pattern(
-    operator: &Operator,
-    path: &str,
-    pattern: Option<Regex>,
-    max_files: usize,
-) -> Result<Vec<StageFileInfo>> {
-    if path == STDIN_FD {
-        return Ok(vec![stdin_stage_info()]);
-    }
-    let operator = operator.blocking();
-    let mut files = Vec::new();
-    let prefix_meta = operator.stat(path);
-    match prefix_meta {
-        Ok(meta) if meta.is_file() => {
-            files.push(StageFileInfo::new(path.to_string(), &meta));
-        }
-        Err(e) if e.kind() != opendal::ErrorKind::NotFound => {
-            return Err(e.into());
-        }
-        _ => {}
-    };
-    let prefix_len = if path == "/" { 0 } else { path.len() };
-    let list = operator.lister_with(path).recursive(true).call()?;
-    if files.len() == max_files {
-        return Ok(files);
-    }
-    for obj in list {
-        let obj = obj?;
-        let (path, mut meta) = obj.into_parts();
-        if check_file(&path[prefix_len..], meta.mode(), &pattern) {
-            if meta.etag().is_none() {
-                meta = match operator.stat(&path) {
-                    Ok(meta) => meta,
-                    Err(err) => return Err(ErrorCode::from(err)),
-                }
-            }
-
-            files.push(StageFileInfo::new(path, &meta));
-            if files.len() == max_files {
-                return Ok(files);
-            }
-        }
-    }
-    Ok(files)
-}
-
 pub const STDIN_FD: &str = "/dev/fd/0";
 
 fn stdin_stage_info() -> StageFileInfo {

src/common/storage/src/stage.rs

@@ -535,12 +535,11 @@ async fn fs_list_until_prefix(
     gc_root_meta_ts: Option<DateTime<Utc>>,
 ) -> Result<Vec<Entry>> {
     // Fetch ALL entries from the path and sort them by path in lexicographical order.
-    let lister = dal.blocking().lister(path)?;
+    let mut lister = dal.lister(path).await?;
     let mut entries = Vec::new();
-    for item in lister {
-        let entry = item?;
-        if entry.metadata().is_file() {
-            entries.push(entry);
+    while let Some(item) = lister.try_next().await? {
+        if item.metadata().is_file() {
+            entries.push(item);
         }
     }
     entries.sort_by(|l, r| l.path().cmp(r.path()));

src/query/ee/src/storages/fuse/operations/vacuum_table_v2.rs

@@ -313,13 +313,9 @@ mod test_accessor {
 
     impl Access for AccessorFaultyDeletion {
         type Reader = ();
-        type BlockingReader = ();
         type Writer = ();
-        type BlockingWriter = ();
         type Lister = VecLister;
-        type BlockingLister = ();
         type Deleter = MockDeleter;
-        type BlockingDeleter = ();
 
         fn info(&self) -> Arc<AccessorInfo> {
             let info = AccessorInfo::default();