This is an automated email from the ASF dual-hosted git repository. xuanwo pushed a commit to branch use-global-client in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit e3fff5ecd51ffa80cf2e13516b882f2948539135 Author: Xuanwo <[email protected]> AuthorDate: Wed Aug 14 17:06:22 2024 +0800 feat: Allow FileIO to reuse http client Signed-off-by: Xuanwo <[email protected]> --- crates/iceberg/src/io/storage.rs | 13 +++++++++++-- crates/iceberg/src/io/storage_s3.rs | 21 ++++++++++++++++----- 2 files changed, 27 insertions(+), 7 deletions(-) diff --git a/crates/iceberg/src/io/storage.rs b/crates/iceberg/src/io/storage.rs index 870e61e..a0a0d76 100644 --- a/crates/iceberg/src/io/storage.rs +++ b/crates/iceberg/src/io/storage.rs @@ -36,6 +36,10 @@ pub(crate) enum Storage { /// s3 storage could have `s3://` and `s3a://`. /// Storing the scheme string here to return the correct path. scheme_str: String, + /// uses the same client for one FileIO Storage. + /// + /// TODO: allow users to configure this client. + client: reqwest::Client, config: Arc<S3Config>, }, } @@ -54,6 +58,7 @@ impl Storage { #[cfg(feature = "storage-s3")] Scheme::S3 => Ok(Self::S3 { scheme_str, + client: reqwest::Client::new(), config: super::s3_config_parse(props)?.into(), }), _ => Err(Error::new( @@ -102,8 +107,12 @@ impl Storage { } } #[cfg(feature = "storage-s3")] - Storage::S3 { scheme_str, config } => { - let op = super::s3_config_build(config, path)?; + Storage::S3 { + scheme_str, + client, + config, + } => { + let op = super::s3_config_build(client, config, path)?; let op_info = op.info(); // Check prefix of s3 path. diff --git a/crates/iceberg/src/io/storage_s3.rs b/crates/iceberg/src/io/storage_s3.rs index 9abb321..4374846 100644 --- a/crates/iceberg/src/io/storage_s3.rs +++ b/crates/iceberg/src/io/storage_s3.rs @@ -17,8 +17,9 @@ use std::collections::HashMap; +use opendal::raw::HttpClient; use opendal::services::S3Config; -use opendal::Operator; +use opendal::{Configurator, Operator}; use url::Url; use crate::{Error, ErrorKind, Result}; @@ -106,7 +107,11 @@ pub(crate) fn s3_config_parse(mut m: HashMap<String, String>) -> Result<S3Config } /// Build new opendal operator from give path. -pub(crate) fn s3_config_build(cfg: &S3Config, path: &str) -> Result<Operator> { +pub(crate) fn s3_config_build( + client: &reqwest::Client, + cfg: &S3Config, + path: &str, +) -> Result<Operator> { let url = Url::parse(path)?; let bucket = url.host_str().ok_or_else(|| { Error::new( @@ -115,7 +120,13 @@ pub(crate) fn s3_config_build(cfg: &S3Config, path: &str) -> Result<Operator> { ) })?; - let mut cfg = cfg.clone(); - cfg.bucket = bucket.to_string(); - Ok(Operator::from_config(cfg)?.finish()) + let builder = cfg + .clone() + .into_builder() + // Set bucket name. + .bucket(bucket) + // Set http client we want to use. + .http_client(HttpClient::with(client.clone())); + + Ok(Operator::new(builder)?.finish()) }
