Merge branch 'main' into fix-doc

Cargo.toml

@@ -93,4 +93,4 @@ arrow-select = { version = "54.2.0", path = "./arrow-select" }
 arrow-string = { version = "54.2.0", path = "./arrow-string" }
 parquet = { version = "54.2.0", path = "./parquet", default-features = false }
 
-chrono = { version = "0.4.34", default-features = false, features = ["clock"] }
+chrono = { version = "0.4.40", default-features = false, features = ["clock"] }

arrow-arith/src/temporal.rs

@@ -658,12 +658,6 @@ pub(crate) use return_compute_error_with;
 
 // Internal trait, which is used for mapping values from DateLike structures
 trait ChronoDateExt {
-    /// Returns a value in range `1..=4` indicating the quarter this date falls into
-    fn quarter(&self) -> u32;
-
-    /// Returns a value in range `0..=3` indicating the quarter (zero-based) this date falls into
-    fn quarter0(&self) -> u32;
-
     /// Returns the day of week; Monday is encoded as `0`, Tuesday as `1`, etc.
     fn num_days_from_monday(&self) -> i32;
 
@@ -672,14 +666,6 @@ trait ChronoDateExt {
 }
 
 impl<T: Datelike> ChronoDateExt for T {
-    fn quarter(&self) -> u32 {
-        self.quarter0() + 1
-    }
-
-    fn quarter0(&self) -> u32 {
-        self.month0() / 3
-    }
-
     fn num_days_from_monday(&self) -> i32 {
         self.weekday().num_days_from_monday() as i32
     }

arrow-data/src/ffi.rs

@@ -289,7 +289,7 @@ impl FFI_ArrowArray {
     /// Returns the buffer at the provided index
     ///
     /// # Panic
-    /// Panics if index exceeds the number of buffers or the buffer is not correctly aligned
+    /// Panics if index >= self.num_buffers() or the buffer is not correctly aligned
     #[inline]
     pub fn buffer(&self, index: usize) -> *const u8 {
         assert!(!self.buffers.is_null());

object_store/Cargo.toml

@@ -45,24 +45,28 @@ walkdir = { version = "2", optional = true }
 
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
+form_urlencoded = { version = "1.2", optional = true }
+http = { version = "1.2.0", optional = true }
+http-body-util = { version = "0.1", optional = true }
+httparse = { version = "1.8.0", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
+md-5 = { version = "0.10.6", default-features = false, optional = true }
 quick-xml = { version = "0.37.0", features = ["serialize", "overlapped-lists"], optional = true }
-serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
-serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "http2"], optional = true }
 ring = { version = "0.17", default-features = false, features = ["std"], optional = true }
 rustls-pemfile = { version = "2.0", default-features = false, features = ["std"], optional = true }
+serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
+serde_json = { version = "1.0", default-features = false, features = ["std"], optional = true }
+serde_urlencoded = { version = "0.7", optional = true }
 tokio = { version = "1.29.0", features = ["sync", "macros", "rt", "time", "io-util"] }
-md-5 = { version = "0.10.6", default-features = false, optional = true }
-httparse = { version = "1.8.0", default-features = false, features = ["std"], optional = true }
 
 [target.'cfg(target_family="unix")'.dev-dependencies]
 nix = { version = "0.29.0", features = ["fs"] }
 
 [features]
 default = ["fs"]
-cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/json", "reqwest/stream", "chrono/serde", "base64", "rand", "ring"]
+cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/stream", "chrono/serde", "base64", "rand", "ring", "dep:http", "http-body-util", "form_urlencoded", "serde_urlencoded"]
 azure = ["cloud", "httparse"]
 fs = ["walkdir"]
 gcp = ["cloud", "rustls-pemfile"]
@@ -72,16 +76,13 @@ tls-webpki-roots = ["reqwest?/rustls-tls-webpki-roots"]
 integration = []
 
 [dev-dependencies] # In alphabetical order
-futures-test = "0.3"
 hyper = { version = "1.2", features = ["server"] }
 hyper-util = "0.1"
-http-body-util = "0.1"
 rand = "0.8"
 tempfile = "3.1.0"
 regex = "1.11.1"
 # The "gzip" feature for reqwest is enabled for an integration test.
 reqwest = { version = "0.12", features = ["gzip"] }
-http = "1.1.0"
 
 [[test]]
 name = "get_range_file"

object_store/src/aws/builder.rs

@@ -23,7 +23,7 @@ use crate::aws::{
     AmazonS3, AwsCredential, AwsCredentialProvider, Checksum, S3ConditionalPut, S3CopyIfNotExists,
     STORE,
 };
-use crate::client::TokenCredentialProvider;
+use crate::client::{HttpConnector, ReqwestConnector, TokenCredentialProvider};
 use crate::config::ConfigValue;
 use crate::{ClientConfigKey, ClientOptions, Result, RetryConfig, StaticCredentialProvider};
 use base64::prelude::BASE64_STANDARD;
@@ -171,6 +171,8 @@ pub struct AmazonS3Builder {
     encryption_customer_key_base64: Option<String>,
     /// When set to true, charge requester for bucket operations
     request_payer: ConfigValue<bool>,
+    /// The [`HttpConnector`] to use
+    http_connector: Option<Arc<dyn HttpConnector>>,
 }
 
 /// Configuration keys for [`AmazonS3Builder`]
@@ -882,13 +884,23 @@ impl AmazonS3Builder {
         self
     }
 
+    /// Overrides the [`HttpConnector`], by default uses [`ReqwestConnector`]
+    pub fn with_http_connector<C: HttpConnector>(mut self, connector: C) -> Self {
+        self.http_connector = Some(Arc::new(connector));
+        self
+    }
+
     /// Create a [`AmazonS3`] instance from the provided values,
     /// consuming `self`.
     pub fn build(mut self) -> Result<AmazonS3> {
         if let Some(url) = self.url.take() {
             self.parse_url(&url)?;
         }
 
+        let http = self
+            .http_connector
+            .unwrap_or_else(|| Arc::new(ReqwestConnector::default()));
+
         let bucket = self.bucket_name.ok_or(Error::MissingBucketName)?;
         let region = self.region.unwrap_or_else(|| "us-east-1".to_string());
         let checksum = self.checksum_algorithm.map(|x| x.get()).transpose()?;
@@ -925,11 +937,7 @@ impl AmazonS3Builder {
             let endpoint = format!("https://sts.{region}.amazonaws.com");
 
             // Disallow non-HTTPs requests
-            let client = self
-                .client_options
-                .clone()
-                .with_allow_http(false)
-                .client()?;
+            let options = self.client_options.clone().with_allow_http(false);
 
             let token = WebIdentityProvider {
                 token_path,
@@ -940,16 +948,19 @@ impl AmazonS3Builder {
 
             Arc::new(TokenCredentialProvider::new(
                 token,
-                client,
+                http.connect(&options)?,
                 self.retry_config.clone(),
             )) as _
         } else if let Some(uri) = self.container_credentials_relative_uri {
             info!("Using Task credential provider");
+
+            let options = self.client_options.clone().with_allow_http(true);
+
             Arc::new(TaskCredentialProvider {
                 url: format!("http://169.254.170.2{uri}"),
                 retry: self.retry_config.clone(),
                 // The instance metadata endpoint is access over HTTP
-                client: self.client_options.clone().with_allow_http(true).client()?,
+                client: http.connect(&options)?,
                 cache: Default::default(),
             }) as _
         } else {
@@ -964,7 +975,7 @@ impl AmazonS3Builder {
 
             Arc::new(TokenCredentialProvider::new(
                 token,
-                self.client_options.metadata_client()?,
+                http.connect(&self.client_options.metadata_options())?,
                 self.retry_config.clone(),
             )) as _
         };
@@ -986,7 +997,7 @@ impl AmazonS3Builder {
                             region: region.clone(),
                             credentials: Arc::clone(&credentials),
                         },
-                        self.client_options.client()?,
+                        http.connect(&self.client_options)?,
                         self.retry_config.clone(),
                     )
                     .with_min_ttl(Duration::from_secs(60)), // Credentials only valid for 5 minutes
@@ -1039,7 +1050,8 @@ impl AmazonS3Builder {
             request_payer: self.request_payer.get()?,
         };
 
-        let client = Arc::new(S3Client::new(config)?);
+        let http_client = http.connect(&config.client_options)?;
+        let client = Arc::new(S3Client::new(config, http_client));
 
         Ok(AmazonS3 { client })
     }