feat: Allow reuse http client in rest catalog (#1221)

## Which issue does this PR close?

Creating a reqwest client is not without cost; it involves DNS caching,
HTTP connection pooling, and TLS certificate loading. The previous
design also made it difficult for users to configure the HTTP client,
such as setting custom timeout values or using a globally shared DNS
resolver.

This PR addressed this by allowing users to create the client and pass
to rest catalog as config.

## What changes are included in this PR?

Allow users to pass a `reqwest::Client` as config.

## Are these changes tested?

Unit tests

Signed-off-by: Xuanwo <[email protected]>

Author: Xuanwo

crates/catalog/rest/src/catalog.rs

@@ -31,7 +31,7 @@ use itertools::Itertools;
 use reqwest::header::{
     HeaderMap, HeaderName, HeaderValue, {self},
 };
-use reqwest::{Method, StatusCode, Url};
+use reqwest::{Client, Method, StatusCode, Url};
 use tokio::sync::OnceCell;
 use typed_builder::TypedBuilder;
 
@@ -58,6 +58,9 @@ pub struct RestCatalogConfig {
 
     #[builder(default)]
     props: HashMap<String, String>,
+
+    #[builder(default)]
+    client: Option<Client>,
 }
 
 impl RestCatalogConfig {
@@ -106,6 +109,11 @@ impl RestCatalogConfig {
         ])
     }
 
+    /// Get the client from the config.
+    pub(crate) fn client(&self) -> Option<Client> {
+        self.client.clone()
+    }
+
     /// Get the token from the config.
     ///
     /// The client can use this token to send requests.

crates/catalog/rest/src/client.rs

@@ -59,9 +59,7 @@ impl HttpClient {
     pub fn new(cfg: &RestCatalogConfig) -> Result<Self> {
         let extra_headers = cfg.extra_headers()?;
         Ok(HttpClient {
-            client: Client::builder()
-                .default_headers(extra_headers.clone())
-                .build()?,
+            client: cfg.client().unwrap_or_default(),
             token: Mutex::new(cfg.token()),
             token_endpoint: cfg.get_token_endpoint(),
             credential: cfg.credential(),
@@ -80,9 +78,7 @@ impl HttpClient {
             .transpose()?
             .unwrap_or(self.extra_headers);
         Ok(HttpClient {
-            client: Client::builder()
-                .default_headers(extra_headers.clone())
-                .build()?,
+            client: cfg.client().unwrap_or(self.client),
             token: Mutex::new(
                 cfg.token()
                     .or_else(|| self.token.into_inner().ok().flatten()),
@@ -162,12 +158,11 @@ impl HttpClient {
         );
 
         let auth_req = self
-            .client
             .request(Method::POST, &self.token_endpoint)
             .form(&params)
             .build()?;
         let auth_url = auth_req.url().clone();
-        let auth_resp = self.client.execute(auth_req).await?;
+        let auth_resp = self.execute(auth_req).await?;
 
         let auth_res: TokenResponse = if auth_resp.status() == StatusCode::OK {
             let text = auth_resp
@@ -220,14 +215,22 @@ impl HttpClient {
 
     #[inline]
     pub fn request<U: IntoUrl>(&self, method: Method, url: U) -> RequestBuilder {
-        self.client.request(method, url)
+        self.client
+            .request(method, url)
+            .headers(self.extra_headers.clone())
+    }
+
+    /// Executes the given `Request` and returns a `Response`.
+    pub async fn execute(&self, mut request: Request) -> Result<Response> {
+        request.headers_mut().extend(self.extra_headers.clone());
+        Ok(self.client.execute(request).await?)
     }
 
     // Queries the Iceberg REST catalog after authentication with the given `Request` and
     // returns a `Response`.
     pub async fn query_catalog(&self, mut request: Request) -> Result<Response> {
         self.authenticate(&mut request).await?;
-        Ok(self.client.execute(request).await?)
+        self.execute(request).await
     }
 }