From 1917adcac1057ee9464ede84cea48247264df191 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Sat, 12 Sep 2026 10:01:17 +0700 Subject: [PATCH 1/5] fix: honor the S3 profile name and file and Hadoop's addressing mode for custom endpoints The profile credentials provider ignored fs.s3a.auth.profile.name and fs.s3a.auth.profile.file, and fs.s3a.path.style.access was applied inverted, so every custom endpoint was addressed path-style whatever the flag said and virtual-hosted addressing was never produced. Carry the profile name and file into the SDK builder, derive the virtual-hosted flag from the path-style setting the way Hadoop does, rebuild the endpoint as bucket.host for virtual-hosted addressing while forcing path-style for IP-literal hosts as the AWS SDK does, and return the effective mode with the endpoint so the two cannot disagree. Closes #4245 Closes #2802 --- docs/source/user-guide/latest/datasources.md | 7 +- native/Cargo.lock | 1 + native/Cargo.toml | 1 + native/core/Cargo.toml | 1 + native/core/src/parquet/objectstore/s3.rs | 674 ++++++++++++++++--- 5 files changed, 603 insertions(+), 81 deletions(-) diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index 97dbece0094..fe52d0552da 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -199,7 +199,7 @@ AWS credential providers can be configured using the `fs.s3a.aws.credentials.pro | `com.amazonaws.auth.InstanceProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider` | Access S3 using EC2 instance metadata service (IMDS) | None | | `com.amazonaws.auth.ContainerCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider`
`com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper` | Access S3 using ECS task credentials | None | | `com.amazonaws.auth.WebIdentityTokenCredentialsProvider`
`software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider` | Authenticate using web identity token file | None | -| `com.amazonaws.auth.profile.ProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider` | Authenticate using a named profile from the local AWS credentials file | None | +| `com.amazonaws.auth.profile.ProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider` | Authenticate using a named profile from the local AWS credentials file | `fs.s3a.auth.profile.name` (optional), `fs.s3a.auth.profile.file` (optional); both apply only when this provider is configured | Multiple credential providers can be specified in a comma-separated list using the `fs.s3a.aws.credentials.provider` configuration, just as Hadoop AWS supports. If `fs.s3a.aws.credentials.provider` is not configured, Hadoop S3A's default credential provider chain will be used. All configuration options also support bucket-specific overrides using the pattern `fs.s3a.bucket.{bucket-name}.{option}`. @@ -216,6 +216,11 @@ Beyond credential providers, Comet's Parquet scan supports additional S3 configu All configuration options support bucket-specific overrides using the pattern `fs.s3a.bucket.{bucket-name}.{option}`. +`fs.s3a.path.style.access` selects how the bucket is placed in the request URL: virtual-hosted +addressing (the default, `false`) sends requests to `https://.`, while path-style +(`true`) sends them to `https:///`, which many S3-compatible services such as MinIO +require. An endpoint whose host is an IP address is always addressed path-style, as the AWS SDK does. + ### S3-Compliant Filesystem Schemes Some environments front an S3-compatible service (MinIO, Ceph RGW, Cloudflare R2, Wasabi, and diff --git a/native/Cargo.lock b/native/Cargo.lock index df5fd4ac14a..d65965913fb 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1963,6 +1963,7 @@ dependencies = [ "async-trait", "aws-config", "aws-credential-types", + "aws-runtime", "base64 0.23.1", "bytes", "comet-contrib-delta", diff --git a/native/Cargo.toml b/native/Cargo.toml index 1805a185e9d..2e86d8cc7f1 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -61,6 +61,7 @@ thiserror = "2" object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"] } url = "2.2" aws-config = "1.8.18" +aws-runtime = "1.9.2" aws-credential-types = "1.2.13" iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "665c64e48e8d33797ecb1a421f327edd9b024879" } iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "665c64e48e8d33797ecb1a421f327edd9b024879", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index 97ce3808cea..a76d6188f33 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -67,6 +67,7 @@ datafusion-comet-shuffle = { workspace = true } object_store = { workspace = true } url = { workspace = true } aws-config = { workspace = true } +aws-runtime = { workspace = true } aws-credential-types = { workspace = true } parking_lot = "0.12.5" # Optional Delta Lake contrib (enabled by the `contrib-delta` feature). Source lives diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index c10fc60816e..29187a3b7e8 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -18,7 +18,7 @@ use log::{debug, error}; use std::collections::HashMap; use std::sync::OnceLock; -use url::Url; +use url::{Host, Url}; use crate::cloud::s3::credential_bridge::{AccessMode, CometS3CredentialBridge}; use crate::execution::jni_api::get_runtime; @@ -34,6 +34,7 @@ use aws_credential_types::{ provider::{error::CredentialsError, ProvideCredentials}, Credentials, }; +use aws_runtime::env_config::file::{EnvConfigFileKind, EnvConfigFiles}; use object_store::{ aws::{AmazonS3Builder, AmazonS3ConfigKey, AwsCredential}, path::Path, @@ -239,24 +240,25 @@ fn extract_s3_config_options( s3_configs.insert(AmazonS3ConfigKey::Region, region.to_string()); } - // Extract and handle path style access (virtual hosted style) - let mut virtual_hosted_style_request = false; - if let Some(path_style) = get_config_trimmed(configs, bucket, "path.style.access") { - virtual_hosted_style_request = path_style.to_lowercase() == "true"; - s3_configs.insert( - AmazonS3ConfigKey::VirtualHostedStyleRequest, - virtual_hosted_style_request.to_string(), - ); - } + // Hadoop defaults fs.s3a.path.style.access to false, which means virtual-hosted addressing, + // and treats non-boolean text as that default. object_store expects the inverse flag. + let path_style_access = get_config_trimmed(configs, bucket, "path.style.access") + .is_some_and(|value| value.eq_ignore_ascii_case("true")); + let mut virtual_hosted_style_request = !path_style_access; - // Extract endpoint configuration and modify if virtual hosted style is enabled + // Extract endpoint configuration and shape it for the selected addressing style. The flag is + // taken from the normalized result so the endpoint and the flag never disagree. if let Some(endpoint) = get_config_trimmed(configs, bucket, "endpoint") { - let normalized_endpoint = - normalize_endpoint(endpoint, bucket, virtual_hosted_style_request); - if let Some(endpoint) = normalized_endpoint { - s3_configs.insert(AmazonS3ConfigKey::Endpoint, endpoint); + if let Some(normalized) = normalize_endpoint(endpoint, bucket, virtual_hosted_style_request) + { + virtual_hosted_style_request = normalized.virtual_hosted_style_request; + s3_configs.insert(AmazonS3ConfigKey::Endpoint, normalized.endpoint); } } + s3_configs.insert( + AmazonS3ConfigKey::VirtualHostedStyleRequest, + virtual_hosted_style_request.to_string(), + ); // Extract request payer configuration if let Some(requester_pays) = get_config_trimmed(configs, bucket, "requester.pays.enabled") { @@ -270,11 +272,21 @@ fn extract_s3_config_options( s3_configs } +/// An endpoint shaped for object_store together with the addressing mode it was shaped for. +#[derive(Debug, Clone, PartialEq)] +struct NormalizedEndpoint { + endpoint: String, + virtual_hosted_style_request: bool, +} + +/// Shapes a Hadoop `fs.s3a.endpoint` value into the endpoint object_store expects: for +/// virtual-hosted requests the bucket becomes the leading host label (`scheme://bucket.host[:port]`), +/// while for path-style requests object_store appends `/bucket` itself so the value passes through. fn normalize_endpoint( endpoint: &str, bucket: &str, virtual_hosted_style_request: bool, -) -> Option { +) -> Option { if endpoint.is_empty() { return None; } @@ -292,15 +304,36 @@ fn normalize_endpoint( endpoint.to_string() }; - if virtual_hosted_style_request { - if endpoint.ends_with("/") { - Some(format!("{endpoint}{bucket}")) - } else { - Some(format!("{endpoint}/{bucket}")) - } - } else { - Some(endpoint) // Avoid extra to_string() call since endpoint is already a String + let path_style = |endpoint: String| { + Some(NormalizedEndpoint { + endpoint, + virtual_hosted_style_request: false, + }) + }; + if !virtual_hosted_style_request { + return path_style(endpoint); } + + // Fall back to the endpoint as written when it cannot be parsed so object_store reports + // the malformed value instead of a mangled one + let Ok(url) = Url::parse(&endpoint) else { + return path_style(endpoint); + }; + // The AWS SDK endpoint rules address IP-literal hosts path-style since `bucket.127.0.0.1` is + // not a valid host. Hadoop does not special-case `localhost`, so neither does this. + let host = match url.host() { + Some(Host::Domain(host)) => host, + _ => return path_style(endpoint), + }; + let port = url + .port() + .map(|port| format!(":{port}")) + .unwrap_or_default(); + let path = url.path().trim_end_matches('/'); + Some(NormalizedEndpoint { + endpoint: format!("{}://{bucket}.{host}{port}{path}", url.scheme()), + virtual_hosted_style_request: true, + }) } fn get_config<'a>( @@ -323,6 +356,17 @@ pub(super) fn get_config_trimmed<'a>( get_config(configs, bucket, property).map(|s| s.trim()) } +/// Like [`get_config_trimmed`] but treats a blank value as unset. +fn get_non_empty_config( + configs: &HashMap, + bucket: &str, + property: &str, +) -> Option { + get_config_trimmed(configs, bucket, property) + .filter(|value| !value.is_empty()) + .map(str::to_string) +} + /// Activation key (without `fs.s3a.` prefix) naming the vendor `CometS3CredentialProvider` FQCN. /// Per-bucket override is honored via [`get_config_trimmed`]. const PROVIDER_CLASS_PROPERTY: &str = "comet.credential.provider.class"; @@ -481,7 +525,10 @@ fn build_aws_credential_provider_metadata( } HADOOP_ASSUMED_ROLE => build_assume_role_credential_provider_metadata(configs, bucket), AWS_WEB_IDENTITY_V1 | AWS_WEB_IDENTITY => Ok(CredentialProviderMetadata::WebIdentity), - AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile), + AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile { + name: get_non_empty_config(configs, bucket, "auth.profile.name"), + file: get_non_empty_config(configs, bucket, "auth.profile.file"), + }), _ => Err(object_store::Error::Generic { store: "S3", source: format!("Unsupported credential provider: {credential_provider_name}").into(), @@ -712,7 +759,10 @@ enum CredentialProviderMetadata { Imds, Environment, WebIdentity, - Profile, + Profile { + name: Option, + file: Option, + }, Static { is_valid: bool, access_key: String, @@ -735,7 +785,7 @@ impl CredentialProviderMetadata { CredentialProviderMetadata::Imds => "Imds", CredentialProviderMetadata::Environment => "Environment", CredentialProviderMetadata::WebIdentity => "WebIdentity", - CredentialProviderMetadata::Profile => "Profile", + CredentialProviderMetadata::Profile { .. } => "Profile", CredentialProviderMetadata::Static { .. } => "Static", CredentialProviderMetadata::AssumeRole { .. } => "AssumeRole", CredentialProviderMetadata::Chain(..) => "Chain", @@ -751,7 +801,17 @@ impl CredentialProviderMetadata { CredentialProviderMetadata::Imds => "Imds".to_string(), CredentialProviderMetadata::Environment => "Environment".to_string(), CredentialProviderMetadata::WebIdentity => "WebIdentity".to_string(), - CredentialProviderMetadata::Profile => "Profile".to_string(), + CredentialProviderMetadata::Profile { name, file } => { + let overrides: Vec = [("name", name), ("file", file)] + .into_iter() + .filter_map(|(key, value)| value.as_ref().map(|v| format!("{key}: {v}"))) + .collect(); + if overrides.is_empty() { + "Profile".to_string() + } else { + format!("Profile({})", overrides.join(", ")) + } + } CredentialProviderMetadata::Static { is_valid, .. } => { format!("Static(valid: {is_valid})") } @@ -814,11 +874,22 @@ impl CredentialProviderMetadata { .build(); Ok(Arc::new(credential_provider)) } - CredentialProviderMetadata::Profile => { - let credential_provider = ProfileFileCredentialsProvider::builder() - .configure(&ProviderConfig::with_default_region().await) - .build(); - Ok(Arc::new(credential_provider)) + CredentialProviderMetadata::Profile { name, file } => { + let mut builder = ProfileFileCredentialsProvider::builder() + .configure(&ProviderConfig::with_default_region().await); + if let Some(name) = name { + builder = builder.profile_name(name); + } + if let Some(file) = file { + // Hadoop's ProfileAWSCredentialsProvider loads fs.s3a.auth.profile.file as a + // credentials-format file and reads nothing else, so mirror that here + builder = builder.profile_files( + EnvConfigFiles::builder() + .with_file(EnvConfigFileKind::Credentials, file) + .build(), + ); + } + Ok(Arc::new(builder.build())) } CredentialProviderMetadata::Static { is_valid, @@ -974,6 +1045,20 @@ mod tests { self } + fn with_property(mut self, property: &str, value: &str) -> Self { + self.configs + .insert(format!("fs.s3a.{property}"), value.to_string()); + self + } + + fn with_bucket_property(mut self, bucket: &str, property: &str, value: &str) -> Self { + self.configs.insert( + format!("fs.s3a.bucket.{bucket}.{property}"), + value.to_string(), + ); + self + } + fn build(self) -> HashMap { self.configs } @@ -995,6 +1080,34 @@ mod tests { ); } + #[test] + #[cfg_attr(miri, ignore)] // AWS credential providers and object_store call foreign functions + fn test_create_store_with_custom_endpoint() { + // object_store must accept the flag and endpoint pair in both addressing modes, and + // create_store enables allow_http so an http endpoint is usable + let url = Url::parse("s3a://test-bucket/comet/data.parquet").unwrap(); + for path_style_access in ["false", "true"] { + let configs = TestConfigBuilder::new() + .with_credential_provider(HADOOP_ANONYMOUS) + .with_region("us-east-1") + .with_property("endpoint", "http://minio.internal:9000") + .with_property("path.style.access", path_style_access) + .build(); + let (_object_store, path) = + create_store(&url, &configs, Duration::from_secs(300)).unwrap(); + assert_eq!(path, Path::from("/comet/data.parquet")); + } + + // An IP-literal endpoint must build without path.style.access being set + let configs = TestConfigBuilder::new() + .with_credential_provider(HADOOP_ANONYMOUS) + .with_region("us-east-1") + .with_property("endpoint", "http://127.0.0.1:9000") + .build(); + let (_object_store, path) = create_store(&url, &configs, Duration::from_secs(300)).unwrap(); + assert_eq!(path, Path::from("/comet/data.parquet")); + } + #[test] fn test_get_config_trimmed() { let configs = TestConfigBuilder::new() @@ -1502,22 +1615,143 @@ mod tests { #[tokio::test] #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions async fn test_profile_credential_provider() { + // (configured name, configured file, expected name, expected file) + let cases = [ + (None, None, None, None), + (Some("analytics"), None, Some("analytics"), None), + ( + None, + Some("/etc/aws/credentials"), + None, + Some("/etc/aws/credentials"), + ), + ( + Some("analytics"), + Some("/etc/aws/credentials"), + Some("analytics"), + Some("/etc/aws/credentials"), + ), + // Empty and blank values are treated as unset, other values are trimmed + (Some(""), Some(" "), None, None), + ( + Some(" analytics "), + Some(" /etc/aws/credentials "), + Some("analytics"), + Some("/etc/aws/credentials"), + ), + ]; for provider_name in [AWS_PROFILE, AWS_PROFILE_V1] { - let configs = TestConfigBuilder::new() - .with_credential_provider(provider_name) - .build(); + for (name, file, expected_name, expected_file) in cases { + let mut builder = TestConfigBuilder::new().with_credential_provider(provider_name); + if let Some(name) = name { + builder = builder.with_property("auth.profile.name", name); + } + if let Some(file) = file { + builder = builder.with_property("auth.profile.file", file); + } + let configs = builder.build(); + + let result = + build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap(); + let test_provider = result + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Profile { + name: expected_name.map(str::to_string), + file: expected_file.map(str::to_string), + }, + "provider {provider_name}, name {name:?}, file {file:?}" + ); + } + } + } - let result = - build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) - .await - .unwrap(); - assert!(result.is_some(), "Should return a credential provider"); + #[tokio::test] + #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions + async fn test_profile_credential_provider_per_bucket_override() { + // Each key is overridden independently, so a bucket can replace just the name or just + // the file while the other key keeps its global value + let configs = TestConfigBuilder::new() + .with_credential_provider(AWS_PROFILE) + .with_property("auth.profile.name", "global-profile") + .with_property("auth.profile.file", "/etc/aws/global-credentials") + .with_bucket_property("name-bucket", "auth.profile.name", "bucket-profile") + .with_bucket_property( + "file-bucket", + "auth.profile.file", + "/etc/aws/bucket-credentials", + ) + .build(); - let test_provider = result.unwrap().metadata(); - assert_eq!(test_provider, CredentialProviderMetadata::Profile); + let cases = [ + ( + "name-bucket", + "bucket-profile", + "/etc/aws/global-credentials", + ), + ( + "file-bucket", + "global-profile", + "/etc/aws/bucket-credentials", + ), + ( + "other-bucket", + "global-profile", + "/etc/aws/global-credentials", + ), + ]; + for (bucket, expected_name, expected_file) in cases { + let result = build_credential_provider(&configs, bucket, Duration::from_secs(300)) + .await + .unwrap(); + let test_provider = result + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Profile { + name: Some(expected_name.to_string()), + file: Some(expected_file.to_string()), + }, + "bucket {bucket}" + ); } } + #[tokio::test] + #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions + async fn test_profile_credential_provider_in_chain() { + let configs = TestConfigBuilder::new() + .with_credential_provider(&format!( + "{AWS_ENVIRONMENT},{AWS_PROFILE},{AWS_INSTANCE_PROFILE}" + )) + .with_property("auth.profile.name", "analytics") + .with_property("auth.profile.file", "/etc/aws/credentials") + .build(); + + let result = build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap(); + let test_provider = result + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Chain(vec![ + CredentialProviderMetadata::Environment, + CredentialProviderMetadata::Profile { + name: Some("analytics".to_string()), + file: Some("/etc/aws/credentials".to_string()), + }, + CredentialProviderMetadata::Imds, + ]) + ); + } + #[tokio::test] #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions async fn test_hadoop_iam_instance_credential_provider() { @@ -1966,75 +2200,335 @@ mod tests { } #[test] - fn test_extract_s3_config_custom_endpoint() { - let cases = vec![ - ("custom.endpoint.com", "https://custom.endpoint.com"), - ("https://custom.endpoint.com", "https://custom.endpoint.com"), + fn test_normalize_endpoint_virtual_hosted_style() { + // Virtual-hosted addressing inserts the bucket as the leading host label. The scheme, + // port and any path suffix are preserved and a trailing slash is dropped. + let cases = [ + ( + "custom.endpoint.com", + "https://test-bucket.custom.endpoint.com", + ), + ( + "http://custom.endpoint.com", + "http://test-bucket.custom.endpoint.com", + ), + ( + "https://custom.endpoint.com/", + "https://test-bucket.custom.endpoint.com", + ), + ( + "http://minio.internal:9000", + "http://test-bucket.minio.internal:9000", + ), + ( + "https://custom.endpoint.com:8443/", + "https://test-bucket.custom.endpoint.com:8443", + ), ( "https://custom.endpoint.com/path/to/resource", - "https://custom.endpoint.com/path/to/resource", + "https://test-bucket.custom.endpoint.com/path/to/resource", + ), + ( + "https://custom.endpoint.com/path/to/resource/", + "https://test-bucket.custom.endpoint.com/path/to/resource", + ), + ( + "s3.us-west-2.amazonaws.com", + "https://test-bucket.s3.us-west-2.amazonaws.com", ), ]; - for (endpoint, configured_endpoint) in cases { - let mut configs = HashMap::new(); - configs.insert("fs.s3a.endpoint".to_string(), endpoint.to_string()); - let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + for (endpoint, expected) in cases { assert_eq!( - s3_configs.get(&AmazonS3ConfigKey::Endpoint), - Some(&configured_endpoint.to_string()) + normalize_endpoint(endpoint, "test-bucket", true), + Some(NormalizedEndpoint { + endpoint: expected.to_string(), + virtual_hosted_style_request: true, + }), + "endpoint {endpoint}" ); } + + assert_eq!( + normalize_endpoint("custom.endpoint.com", "my.dotted.bucket", true), + Some(NormalizedEndpoint { + endpoint: "https://my.dotted.bucket.custom.endpoint.com".to_string(), + virtual_hosted_style_request: true, + }) + ); } #[test] - fn test_extract_s3_config_custom_endpoint_with_virtual_hosted_style() { - let cases = vec![ + fn test_normalize_endpoint_path_style() { + // Path-style leaves the endpoint as configured apart from the https default, since + // object_store appends the bucket itself + let cases = [ + ("custom.endpoint.com", "https://custom.endpoint.com"), + ("http://custom.endpoint.com", "http://custom.endpoint.com"), ( - "custom.endpoint.com", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com/", + "https://custom.endpoint.com/", ), + ("http://minio.internal:9000", "http://minio.internal:9000"), ( - "https://custom.endpoint.com", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com:8443/", + "https://custom.endpoint.com:8443/", ), ( - "https://custom.endpoint.com/", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com/path/to/resource", + "https://custom.endpoint.com/path/to/resource", ), ( - "https://custom.endpoint.com/path/to/resource", - "https://custom.endpoint.com/path/to/resource/test-bucket", + "s3.us-west-2.amazonaws.com", + "https://s3.us-west-2.amazonaws.com", ), + ]; + for (endpoint, expected) in cases { + assert_eq!( + normalize_endpoint(endpoint, "test-bucket", false), + Some(NormalizedEndpoint { + endpoint: expected.to_string(), + virtual_hosted_style_request: false, + }), + "endpoint {endpoint}" + ); + } + + assert_eq!( + normalize_endpoint("custom.endpoint.com", "my.dotted.bucket", false), + Some(NormalizedEndpoint { + endpoint: "https://custom.endpoint.com".to_string(), + virtual_hosted_style_request: false, + }) + ); + } + + #[test] + fn test_normalize_endpoint_ip_host_forces_path_style() { + // The AWS SDK endpoint rules address IP-literal hosts path-style whatever the + // configuration says, since `bucket.127.0.0.1` is not a valid host + let cases = [ + ("http://127.0.0.1:9000", "http://127.0.0.1:9000"), + ("http://127.0.0.1", "http://127.0.0.1"), + ("127.0.0.1:9000", "https://127.0.0.1:9000"), + ("http://[::1]:9000", "http://[::1]:9000"), + ("https://[::1]", "https://[::1]"), + ("[::1]:9000", "https://[::1]:9000"), + ]; + for (endpoint, expected) in cases { + for virtual_hosted_style_request in [true, false] { + assert_eq!( + normalize_endpoint(endpoint, "test-bucket", virtual_hosted_style_request), + Some(NormalizedEndpoint { + endpoint: expected.to_string(), + virtual_hosted_style_request: false, + }), + "endpoint {endpoint}, requested virtual-hosted {virtual_hosted_style_request}" + ); + } + } + } + + #[test] + fn test_normalize_endpoint_skips_default_aws_endpoint() { + for virtual_hosted_style_request in [true, false] { + assert_eq!( + normalize_endpoint( + "s3.amazonaws.com", + "test-bucket", + virtual_hosted_style_request + ), + None + ); + assert_eq!( + normalize_endpoint("", "test-bucket", virtual_hosted_style_request), + None + ); + } + } + + #[test] + fn test_extract_s3_config_path_style_access() { + // Hadoop defaults fs.s3a.path.style.access to false (virtual-hosted) and, like + // Configuration.getBoolean, falls back to that default for non-boolean text + let cases = [ + (None, "true", "https://test-bucket.custom.endpoint.com"), ( - "https://custom.endpoint.com/path/to/resource/", - "https://custom.endpoint.com/path/to/resource/test-bucket", + Some("false"), + "true", + "https://test-bucket.custom.endpoint.com", + ), + ( + Some("yes"), + "true", + "https://test-bucket.custom.endpoint.com", ), + (Some("true"), "false", "https://custom.endpoint.com"), + (Some(" TRUE "), "false", "https://custom.endpoint.com"), ]; - for (endpoint, configured_endpoint) in cases { - let mut configs = HashMap::new(); - configs.insert("fs.s3a.endpoint".to_string(), endpoint.to_string()); - configs.insert("fs.s3a.path.style.access".to_string(), "true".to_string()); + for (path_style_access, expected_flag, expected_endpoint) in cases { + let mut builder = + TestConfigBuilder::new().with_property("endpoint", "custom.endpoint.com"); + if let Some(value) = path_style_access { + builder = builder.with_property("path.style.access", value); + } + let s3_configs = extract_s3_config_options(&builder.build(), "test-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&expected_flag.to_string()), + "path.style.access {path_style_access:?}" + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&expected_endpoint.to_string()), + "path.style.access {path_style_access:?}" + ); + } + } + + #[test] + fn test_extract_s3_config_ip_endpoint_forces_path_style() { + // With path.style.access unset an IP-literal endpoint stays path-style and the flag + // handed to object_store agrees with the unchanged endpoint + for endpoint in [ + "http://127.0.0.1:9000", + "http://127.0.0.1", + "http://[::1]:9000", + "http://[::1]", + ] { + let configs = TestConfigBuilder::new() + .with_property("endpoint", endpoint) + .build(); let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()), + "endpoint {endpoint}" + ); assert_eq!( s3_configs.get(&AmazonS3ConfigKey::Endpoint), - Some(&configured_endpoint.to_string()) + Some(&endpoint.to_string()), + "endpoint {endpoint}" ); } } #[test] - fn test_extract_s3_config_ignore_default_endpoint() { - let mut configs = HashMap::new(); - configs.insert( - "fs.s3a.endpoint".to_string(), - "s3.amazonaws.com".to_string(), - ); + fn test_extract_s3_config_http_endpoint_keeps_scheme() { + for (path_style_access, expected_endpoint) in [ + ("false", "http://test-bucket.minio.internal:9000"), + ("true", "http://minio.internal:9000"), + ] { + let configs = TestConfigBuilder::new() + .with_property("endpoint", "http://minio.internal:9000") + .with_property("path.style.access", path_style_access) + .build(); + let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&expected_endpoint.to_string()), + "path.style.access {path_style_access}" + ); + } + } + + #[test] + fn test_extract_s3_config_path_style_access_without_endpoint() { + // The flag is always handed to object_store so the default AWS endpoint follows the + // same addressing rule as a custom one + let configs = TestConfigBuilder::new().with_region("us-east-1").build(); let s3_configs = extract_s3_config_options(&configs, "test-bucket"); - assert!(s3_configs.is_empty()); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); - configs.insert("fs.s3a.endpoint".to_string(), "".to_string()); + let configs = TestConfigBuilder::new() + .with_region("us-east-1") + .with_property("path.style.access", "true") + .build(); let s3_configs = extract_s3_config_options(&configs, "test-bucket"); - assert!(s3_configs.is_empty()); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + } + + #[test] + fn test_extract_s3_config_per_bucket_overrides() { + // A bucket can override both the endpoint and the addressing flag, in either direction + let configs = TestConfigBuilder::new() + .with_property("endpoint", "global.endpoint.com") + .with_property("path.style.access", "true") + .with_bucket_property("vh-bucket", "endpoint", "http://bucket.endpoint.com:9000") + .with_bucket_property("vh-bucket", "path.style.access", "false") + .build(); + + let s3_configs = extract_s3_config_options(&configs, "vh-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"http://vh-bucket.bucket.endpoint.com:9000".to_string()) + ); + + let s3_configs = extract_s3_config_options(&configs, "other-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://global.endpoint.com".to_string()) + ); + + let configs = TestConfigBuilder::new() + .with_property("endpoint", "global.endpoint.com") + .with_property("path.style.access", "false") + .with_bucket_property("ps-bucket", "path.style.access", "true") + .build(); + + let s3_configs = extract_s3_config_options(&configs, "ps-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://global.endpoint.com".to_string()) + ); + + let s3_configs = extract_s3_config_options(&configs, "other-bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://other-bucket.global.endpoint.com".to_string()) + ); + } + + #[test] + fn test_extract_s3_config_ignore_default_endpoint() { + for path_style_access in ["false", "true"] { + let configs = TestConfigBuilder::new() + .with_property("endpoint", "s3.amazonaws.com") + .with_property("path.style.access", path_style_access) + .build(); + let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + + let configs = TestConfigBuilder::new() + .with_property("endpoint", "") + .with_property("path.style.access", path_style_access) + .build(); + let s3_configs = extract_s3_config_options(&configs, "test-bucket"); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + } } #[test] @@ -2059,6 +2553,26 @@ mod tests { "AssumeRole(role: arn:aws:iam::123456789012:role/test-role, session: test-session, base: Environment)" ); + // Test Profile provider with and without overrides + let profile_metadata = CredentialProviderMetadata::Profile { + name: None, + file: None, + }; + assert_eq!(profile_metadata.simple_string(), "Profile"); + let profile_metadata = CredentialProviderMetadata::Profile { + name: Some("analytics".to_string()), + file: None, + }; + assert_eq!(profile_metadata.simple_string(), "Profile(name: analytics)"); + let profile_metadata = CredentialProviderMetadata::Profile { + name: Some("analytics".to_string()), + file: Some("/etc/aws/credentials".to_string()), + }; + assert_eq!( + profile_metadata.simple_string(), + "Profile(name: analytics, file: /etc/aws/credentials)" + ); + // Test Chain provider let chain_metadata = CredentialProviderMetadata::Chain(vec![ CredentialProviderMetadata::Static { From 80e1eb90904d277d72c283261c510d45ce9a1754 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Mon, 14 Sep 2026 21:13:10 +0700 Subject: [PATCH 2/5] fix: apply the S3 profile keys only to Hadoop's profile provider and document the addressing change --- docs/source/user-guide/latest/datasources.md | 6 ++- native/core/src/parquet/objectstore/s3.rs | 47 +++++++++++++++++--- 2 files changed, 47 insertions(+), 6 deletions(-) diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index fe52d0552da..ce0ed721899 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -199,7 +199,8 @@ AWS credential providers can be configured using the `fs.s3a.aws.credentials.pro | `com.amazonaws.auth.InstanceProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider` | Access S3 using EC2 instance metadata service (IMDS) | None | | `com.amazonaws.auth.ContainerCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider`
`com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper` | Access S3 using ECS task credentials | None | | `com.amazonaws.auth.WebIdentityTokenCredentialsProvider`
`software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider` | Authenticate using web identity token file | None | -| `com.amazonaws.auth.profile.ProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider` | Authenticate using a named profile from the local AWS credentials file | `fs.s3a.auth.profile.name` (optional), `fs.s3a.auth.profile.file` (optional); both apply only when this provider is configured | +| `org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider` | Authenticate using a named profile from the local AWS credentials file | `fs.s3a.auth.profile.name` (optional), `fs.s3a.auth.profile.file` (optional); Hadoop applies both only to this provider | +| `com.amazonaws.auth.profile.ProfileCredentialsProvider`
`software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider` | Authenticate using the SDK's default profile; Hadoop constructs these without its configuration, so the profile keys are not applied on either side | None | Multiple credential providers can be specified in a comma-separated list using the `fs.s3a.aws.credentials.provider` configuration, just as Hadoop AWS supports. If `fs.s3a.aws.credentials.provider` is not configured, Hadoop S3A's default credential provider chain will be used. All configuration options also support bucket-specific overrides using the pattern `fs.s3a.bucket.{bucket-name}.{option}`. @@ -220,6 +221,9 @@ All configuration options support bucket-specific overrides using the pattern `f addressing (the default, `false`) sends requests to `https://.`, while path-style (`true`) sends them to `https:///`, which many S3-compatible services such as MinIO require. An endpoint whose host is an IP address is always addressed path-style, as the AWS SDK does. +Earlier Comet releases addressed every custom `fs.s3a.endpoint` path-style whatever the flag said, +so a MinIO or Ceph RGW deployment that never set the flag now sends requests to `.` +and fails with a DNS error; set `fs.s3a.path.style.access=true` to keep the previous behavior. ### S3-Compliant Filesystem Schemes diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index 29187a3b7e8..81a92be2fc4 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -402,6 +402,7 @@ const AWS_WEB_IDENTITY: &str = const AWS_WEB_IDENTITY_V1: &str = "com.amazonaws.auth.WebIdentityTokenCredentialsProvider"; const AWS_PROFILE: &str = "software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider"; const AWS_PROFILE_V1: &str = "com.amazonaws.auth.profile.ProfileCredentialsProvider"; +const HADOOP_PROFILE: &str = "org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider"; const AWS_ANONYMOUS: &str = "software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider"; const AWS_ANONYMOUS_V1: &str = "com.amazonaws.auth.AnonymousAWSCredentials"; @@ -525,10 +526,17 @@ fn build_aws_credential_provider_metadata( } HADOOP_ASSUMED_ROLE => build_assume_role_credential_provider_metadata(configs, bucket), AWS_WEB_IDENTITY_V1 | AWS_WEB_IDENTITY => Ok(CredentialProviderMetadata::WebIdentity), - AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile { + // Only Hadoop's own provider reads the profile keys. Hadoop builds the SDK spellings + // through the SDK's static constructor without its configuration, so applying the keys + // to them here would authenticate the native side as a different identity. + HADOOP_PROFILE => Ok(CredentialProviderMetadata::Profile { name: get_non_empty_config(configs, bucket, "auth.profile.name"), file: get_non_empty_config(configs, bucket, "auth.profile.file"), }), + AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile { + name: None, + file: None, + }), _ => Err(object_store::Error::Generic { store: "S3", source: format!("Unsupported credential provider: {credential_provider_name}").into(), @@ -1640,8 +1648,9 @@ mod tests { Some("/etc/aws/credentials"), ), ]; - for provider_name in [AWS_PROFILE, AWS_PROFILE_V1] { - for (name, file, expected_name, expected_file) in cases { + for (name, file, expected_name, expected_file) in cases { + { + let provider_name = HADOOP_PROFILE; let mut builder = TestConfigBuilder::new().with_credential_provider(provider_name); if let Some(name) = name { builder = builder.with_property("auth.profile.name", name); @@ -1670,13 +1679,41 @@ mod tests { } } + #[tokio::test] + #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions + async fn test_sdk_profile_provider_spellings_ignore_the_profile_keys() { + // Hadoop constructs these spellings without its configuration, so the native side must + // resolve the SDK default profile too, even when the keys are set. + for provider_name in [AWS_PROFILE, AWS_PROFILE_V1] { + let configs = TestConfigBuilder::new() + .with_credential_provider(provider_name) + .with_property("auth.profile.name", "analytics") + .with_property("auth.profile.file", "/etc/aws/credentials") + .build(); + let test_provider = + build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap() + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Profile { + name: None, + file: None, + }, + "provider {provider_name}" + ); + } + } + #[tokio::test] #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions async fn test_profile_credential_provider_per_bucket_override() { // Each key is overridden independently, so a bucket can replace just the name or just // the file while the other key keeps its global value let configs = TestConfigBuilder::new() - .with_credential_provider(AWS_PROFILE) + .with_credential_provider(HADOOP_PROFILE) .with_property("auth.profile.name", "global-profile") .with_property("auth.profile.file", "/etc/aws/global-credentials") .with_bucket_property("name-bucket", "auth.profile.name", "bucket-profile") @@ -1727,7 +1764,7 @@ mod tests { async fn test_profile_credential_provider_in_chain() { let configs = TestConfigBuilder::new() .with_credential_provider(&format!( - "{AWS_ENVIRONMENT},{AWS_PROFILE},{AWS_INSTANCE_PROFILE}" + "{AWS_ENVIRONMENT},{HADOOP_PROFILE},{AWS_INSTANCE_PROFILE}" )) .with_property("auth.profile.name", "analytics") .with_property("auth.profile.file", "/etc/aws/credentials") From 0c76e6c1aaf65ebd23e5f4d3cfb76a3d42644f97 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Tue, 15 Sep 2026 16:51:32 +0700 Subject: [PATCH 3/5] fix: keep dotted buckets path-style over HTTPS and read only the credentials file for Hadoop's profile provider --- docs/source/user-guide/latest/datasources.md | 4 +- native/core/src/parquet/objectstore/s3.rs | 124 +++++++++++++++++-- 2 files changed, 120 insertions(+), 8 deletions(-) diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index adabfd55298..7d7a21a966e 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -242,7 +242,9 @@ All configuration options support bucket-specific overrides using the pattern `f `fs.s3a.path.style.access` selects how the bucket is placed in the request URL: virtual-hosted addressing (the default, `false`) sends requests to `https://.`, while path-style (`true`) sends them to `https:///`, which many S3-compatible services such as MinIO -require. An endpoint whose host is an IP address is always addressed path-style, as the AWS SDK does. +require. An endpoint whose host is an IP address is always addressed path-style, as the AWS SDK does, +and so is a bucket whose name contains a dot over HTTPS, since the dotted host falls outside S3's +wildcard certificate. Earlier Comet releases addressed every custom `fs.s3a.endpoint` path-style whatever the flag said, so a MinIO or Ceph RGW deployment that never set the flag now sends requests to `.` and fails with a DNS error; set `fs.s3a.path.style.access=true` to keep the previous behavior. diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index 81a92be2fc4..17cec047e24 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -244,7 +244,11 @@ fn extract_s3_config_options( // and treats non-boolean text as that default. object_store expects the inverse flag. let path_style_access = get_config_trimmed(configs, bucket, "path.style.access") .is_some_and(|value| value.eq_ignore_ascii_case("true")); - let mut virtual_hosted_style_request = !path_style_access; + // The AWS SDK addresses a bucket whose name contains a dot path-style over HTTPS, because + // the dotted host does not match S3's wildcard certificate. The default AWS endpoint is + // HTTPS, and normalize_endpoint applies the same rule to a custom one by its scheme. + let mut virtual_hosted_style_request = + !path_style_access && !bucket_needs_path_style_over_https(bucket); // Extract endpoint configuration and shape it for the selected addressing style. The flag is // taken from the normalized result so the endpoint and the flag never disagree. @@ -272,6 +276,12 @@ fn extract_s3_config_options( s3_configs } +/// Whether the AWS SDK would refuse to virtual-host `bucket` over HTTPS: a dot in the name +/// makes `bucket.s3..amazonaws.com` fall outside S3's wildcard certificate. +fn bucket_needs_path_style_over_https(bucket: &str) -> bool { + bucket.contains('.') +} + /// An endpoint shaped for object_store together with the addressing mode it was shaped for. #[derive(Debug, Clone, PartialEq)] struct NormalizedEndpoint { @@ -313,6 +323,9 @@ fn normalize_endpoint( if !virtual_hosted_style_request { return path_style(endpoint); } + if endpoint.starts_with("https://") && bucket_needs_path_style_over_https(bucket) { + return path_style(endpoint); + } // Fall back to the endpoint as written when it cannot be parsed so object_store reports // the malformed value instead of a mangled one @@ -336,6 +349,14 @@ fn normalize_endpoint( }) } +/// The credentials file Hadoop's profile provider reads when none is configured: +/// `AWS_SHARED_CREDENTIALS_FILE` when set, otherwise `~/.aws/credentials`. +fn default_shared_credentials_file(env_override: Option, home: Option) -> String { + env_override + .filter(|path| !path.trim().is_empty()) + .unwrap_or_else(|| format!("{}/.aws/credentials", home.unwrap_or_default())) +} + fn get_config<'a>( configs: &'a HashMap, bucket: &str, @@ -532,10 +553,12 @@ fn build_aws_credential_provider_metadata( HADOOP_PROFILE => Ok(CredentialProviderMetadata::Profile { name: get_non_empty_config(configs, bucket, "auth.profile.name"), file: get_non_empty_config(configs, bucket, "auth.profile.file"), + credentials_only: true, }), AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile { name: None, file: None, + credentials_only: false, }), _ => Err(object_store::Error::Generic { store: "S3", @@ -770,6 +793,9 @@ enum CredentialProviderMetadata { Profile { name: Option, file: Option, + // Hadoop's ProfileAWSCredentialsProvider reads only the credentials file, while the + // SDK spellings merge the SDK's config and credentials files. + credentials_only: bool, }, Static { is_valid: bool, @@ -809,7 +835,7 @@ impl CredentialProviderMetadata { CredentialProviderMetadata::Imds => "Imds".to_string(), CredentialProviderMetadata::Environment => "Environment".to_string(), CredentialProviderMetadata::WebIdentity => "WebIdentity".to_string(), - CredentialProviderMetadata::Profile { name, file } => { + CredentialProviderMetadata::Profile { name, file, .. } => { let overrides: Vec = [("name", name), ("file", file)] .into_iter() .filter_map(|(key, value)| value.as_ref().map(|v| format!("{key}: {v}"))) @@ -882,15 +908,28 @@ impl CredentialProviderMetadata { .build(); Ok(Arc::new(credential_provider)) } - CredentialProviderMetadata::Profile { name, file } => { + CredentialProviderMetadata::Profile { + name, + file, + credentials_only, + } => { let mut builder = ProfileFileCredentialsProvider::builder() .configure(&ProviderConfig::with_default_region().await); if let Some(name) = name { builder = builder.profile_name(name); } - if let Some(file) = file { - // Hadoop's ProfileAWSCredentialsProvider loads fs.s3a.auth.profile.file as a - // credentials-format file and reads nothing else, so mirror that here + // Hadoop's ProfileAWSCredentialsProvider loads the configured file, or the + // shared credentials file, as a credentials-format file and reads nothing + // else, so a same-name role profile in the SDK's config file never applies. + let credentials_file = match (file, credentials_only) { + (Some(file), _) => Some(file.clone()), + (None, true) => Some(default_shared_credentials_file( + std::env::var("AWS_SHARED_CREDENTIALS_FILE").ok(), + std::env::var("HOME").ok(), + )), + (None, false) => None, + }; + if let Some(file) = credentials_file { builder = builder.profile_files( EnvConfigFiles::builder() .with_file(EnvConfigFileKind::Credentials, file) @@ -1672,6 +1711,7 @@ mod tests { CredentialProviderMetadata::Profile { name: expected_name.map(str::to_string), file: expected_file.map(str::to_string), + credentials_only: true, }, "provider {provider_name}, name {name:?}, file {file:?}" ); @@ -1701,6 +1741,7 @@ mod tests { CredentialProviderMetadata::Profile { name: None, file: None, + credentials_only: false, }, "provider {provider_name}" ); @@ -1753,6 +1794,7 @@ mod tests { CredentialProviderMetadata::Profile { name: Some(expected_name.to_string()), file: Some(expected_file.to_string()), + credentials_only: true, }, "bucket {bucket}" ); @@ -1783,6 +1825,7 @@ mod tests { CredentialProviderMetadata::Profile { name: Some("analytics".to_string()), file: Some("/etc/aws/credentials".to_string()), + credentials_only: true, }, CredentialProviderMetadata::Imds, ]) @@ -2285,15 +2328,79 @@ mod tests { ); } + // A dotted bucket over HTTPS stays path-style, as the AWS SDK addresses it, since the + // dotted host falls outside S3's wildcard certificate; over HTTP it is virtual-hosted. assert_eq!( normalize_endpoint("custom.endpoint.com", "my.dotted.bucket", true), Some(NormalizedEndpoint { - endpoint: "https://my.dotted.bucket.custom.endpoint.com".to_string(), + endpoint: "https://custom.endpoint.com".to_string(), + virtual_hosted_style_request: false, + }) + ); + assert_eq!( + normalize_endpoint("http://custom.endpoint.com", "my.dotted.bucket", true), + Some(NormalizedEndpoint { + endpoint: "http://my.dotted.bucket.custom.endpoint.com".to_string(), virtual_hosted_style_request: true, }) ); } + #[test] + fn test_extract_s3_config_dotted_bucket_stays_path_style_on_default_endpoint() { + // No custom endpoint means the HTTPS AWS endpoint, where a dotted bucket must be + // addressed path-style whatever the flag says. + let configs = TestConfigBuilder::new().with_region("us-east-1").build(); + let s3_configs = extract_s3_config_options(&configs, "review.dotted.bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert!(!s3_configs.contains_key(&AmazonS3ConfigKey::Endpoint)); + + let configs = TestConfigBuilder::new() + .with_region("us-east-1") + .with_property("endpoint", "https://s3.us-east-1.amazonaws.com") + .build(); + let s3_configs = extract_s3_config_options(&configs, "review.dotted.bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"false".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"https://s3.us-east-1.amazonaws.com".to_string()) + ); + + let s3_configs = extract_s3_config_options(&configs, "plainbucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + } + + #[test] + fn test_default_shared_credentials_file_matches_hadoop() { + assert_eq!( + default_shared_credentials_file(None, Some("/home/comet".to_string())), + "/home/comet/.aws/credentials" + ); + assert_eq!( + default_shared_credentials_file( + Some("/etc/aws/shared".to_string()), + Some("/home/comet".to_string()) + ), + "/etc/aws/shared" + ); + assert_eq!( + default_shared_credentials_file( + Some(" ".to_string()), + Some("/home/comet".to_string()) + ), + "/home/comet/.aws/credentials" + ); + } + #[test] fn test_normalize_endpoint_path_style() { // Path-style leaves the endpoint as configured apart from the https default, since @@ -2594,16 +2701,19 @@ mod tests { let profile_metadata = CredentialProviderMetadata::Profile { name: None, file: None, + credentials_only: false, }; assert_eq!(profile_metadata.simple_string(), "Profile"); let profile_metadata = CredentialProviderMetadata::Profile { name: Some("analytics".to_string()), file: None, + credentials_only: true, }; assert_eq!(profile_metadata.simple_string(), "Profile(name: analytics)"); let profile_metadata = CredentialProviderMetadata::Profile { name: Some("analytics".to_string()), file: Some("/etc/aws/credentials".to_string()), + credentials_only: true, }; assert_eq!( profile_metadata.simple_string(), From 31fafaffd01e820bb212de91a73ae8f0b63a6e45 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Tue, 15 Sep 2026 23:27:19 +0700 Subject: [PATCH 4/5] fix: decide dotted-bucket addressing by the endpoint scheme and forward Hadoop's default credentials path from the JVM --- native/core/src/parquet/objectstore/s3.rs | 99 +++++++++++++++++-- .../comet/objectstore/NativeConfig.scala | 22 +++++ .../comet/objectstore/NativeConfigSuite.scala | 29 ++++++ 3 files changed, 140 insertions(+), 10 deletions(-) diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index 17cec047e24..26b3bd27f89 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -244,20 +244,24 @@ fn extract_s3_config_options( // and treats non-boolean text as that default. object_store expects the inverse flag. let path_style_access = get_config_trimmed(configs, bucket, "path.style.access") .is_some_and(|value| value.eq_ignore_ascii_case("true")); - // The AWS SDK addresses a bucket whose name contains a dot path-style over HTTPS, because - // the dotted host does not match S3's wildcard certificate. The default AWS endpoint is - // HTTPS, and normalize_endpoint applies the same rule to a custom one by its scheme. - let mut virtual_hosted_style_request = - !path_style_access && !bucket_needs_path_style_over_https(bucket); + let mut virtual_hosted_style_request = !path_style_access; // Extract endpoint configuration and shape it for the selected addressing style. The flag is - // taken from the normalized result so the endpoint and the flag never disagree. - if let Some(endpoint) = get_config_trimmed(configs, bucket, "endpoint") { - if let Some(normalized) = normalize_endpoint(endpoint, bucket, virtual_hosted_style_request) - { + // taken from the normalized result so the endpoint and the flag never disagree. A custom + // endpoint decides the dotted-bucket rule by its own scheme inside normalize_endpoint; the + // default AWS endpoint is HTTPS, so the rule applies to it here. + let custom_endpoint = get_config_trimmed(configs, bucket, "endpoint") + .and_then(|endpoint| normalize_endpoint(endpoint, bucket, virtual_hosted_style_request)); + match custom_endpoint { + Some(normalized) => { virtual_hosted_style_request = normalized.virtual_hosted_style_request; s3_configs.insert(AmazonS3ConfigKey::Endpoint, normalized.endpoint); } + None => { + if bucket_needs_path_style_over_https(bucket) { + virtual_hosted_style_request = false; + } + } } s3_configs.insert( AmazonS3ConfigKey::VirtualHostedStyleRequest, @@ -550,9 +554,13 @@ fn build_aws_credential_provider_metadata( // Only Hadoop's own provider reads the profile keys. Hadoop builds the SDK spellings // through the SDK's static constructor without its configuration, so applying the keys // to them here would authenticate the native side as a different identity. + // With no configured file, Hadoop reads AWS_SHARED_CREDENTIALS_FILE or the JVM + // user's ~/.aws/credentials; the JVM forwards that resolved path so both sides agree + // even when the native process sees a different HOME. HADOOP_PROFILE => Ok(CredentialProviderMetadata::Profile { name: get_non_empty_config(configs, bucket, "auth.profile.name"), - file: get_non_empty_config(configs, bucket, "auth.profile.file"), + file: get_non_empty_config(configs, bucket, "auth.profile.file") + .or_else(|| get_non_empty_config(configs, bucket, "comet.default.profile.file")), credentials_only: true, }), AWS_PROFILE_V1 | AWS_PROFILE => Ok(CredentialProviderMetadata::Profile { @@ -2377,6 +2385,77 @@ mod tests { s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), Some(&"true".to_string()) ); + + // A custom HTTP endpoint keeps virtual hosting for a dotted bucket, as the SDK does, + // since the certificate rule only applies to HTTPS. + let configs = TestConfigBuilder::new() + .with_property("endpoint", "http://storage.example.test") + .build(); + let s3_configs = extract_s3_config_options(&configs, "review.dotted.bucket"); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::VirtualHostedStyleRequest), + Some(&"true".to_string()) + ); + assert_eq!( + s3_configs.get(&AmazonS3ConfigKey::Endpoint), + Some(&"http://review.dotted.bucket.storage.example.test".to_string()) + ); + } + + #[tokio::test] + #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions + async fn test_hadoop_profile_provider_takes_the_forwarded_default_file() { + // With no configured file the JVM-resolved default applies; a configured file wins; + // the SDK spellings ignore both. + for (file, expected) in [ + (None, Some("/synthetic/jvm-home/.aws/credentials")), + (Some("/etc/aws/credentials"), Some("/etc/aws/credentials")), + ] { + let mut builder = TestConfigBuilder::new() + .with_credential_provider(HADOOP_PROFILE) + .with_property( + "comet.default.profile.file", + "/synthetic/jvm-home/.aws/credentials", + ); + if let Some(file) = file { + builder = builder.with_property("auth.profile.file", file); + } + let configs = builder.build(); + let metadata = + build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap() + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + metadata, + CredentialProviderMetadata::Profile { + name: None, + file: expected.map(str::to_string), + credentials_only: true, + } + ); + } + let configs = TestConfigBuilder::new() + .with_credential_provider(AWS_PROFILE) + .with_property( + "comet.default.profile.file", + "/synthetic/jvm-home/.aws/credentials", + ) + .build(); + let metadata = build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) + .await + .unwrap() + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + metadata, + CredentialProviderMetadata::Profile { + name: None, + file: None, + credentials_only: false, + } + ); } #[test] diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index 75328649fea..28cac3e478c 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -196,6 +196,21 @@ object NativeConfig { * * The result feeds object_store's parse_url_opts natively. */ + /** + * Where the native side finds the credentials file Hadoop's profile provider would read with no + * `fs.s3a.auth.profile.file` configured: `AWS_SHARED_CREDENTIALS_FILE`, else the JVM user's + * `~/.aws/credentials` (Hadoop resolves the home through `user.home`, not `HOME`). + */ + val COMET_DEFAULT_PROFILE_FILE_KEY = "fs.s3a.comet.default.profile.file" + + private[objectstore] def defaultSharedCredentialsFile( + env: Map[String, String] = sys.env, + userHome: String = System.getProperty("user.home")): String = + env + .get("AWS_SHARED_CREDENTIALS_FILE") + .filter(StringUtils.isNotBlank) + .getOrElse(new java.io.File(new java.io.File(userHome, ".aws"), "credentials").getPath) + def extractObjectStoreOptions(hadoopConf: Configuration, uri: URI): Map[String, String] = { val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("file") @@ -231,6 +246,13 @@ object NativeConfig { val vendorPrefix = if (s3CompliantSchemes.contains(scheme)) s"fs.$scheme." else "" val vendorEntries = scala.collection.mutable.ArrayBuffer[(String, String)]() + // Hadoop's ProfileAWSCredentialsProvider reads this file when fs.s3a.auth.profile.file is + // unset; native resolves paths against its own environment, so the JVM's answer rides along + // for every scheme that uses the fs.s3a.* surface. + if (prefixes.get.contains("fs.s3a.")) { + options(COMET_DEFAULT_PROFILE_FILE_KEY) = defaultSharedCredentialsFile() + } + hadoopConf.iterator().asScala.foreach { entry => val key = entry.getKey if (prefixes.get.exists(prefix => key.startsWith(prefix))) { diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index 8e5a60d63b0..239d0a56b6a 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -30,6 +30,35 @@ import org.apache.comet.CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY class NativeConfigSuite extends AnyFunSuite with Matchers { + test("extractObjectStoreOptions forwards the JVM-resolved default credentials file") { + val hadoopConf = new Configuration() + val options = + NativeConfig.extractObjectStoreOptions(hadoopConf, new URI("s3a://test-bucket/object")) + val expected = new java.io.File( + new java.io.File(System.getProperty("user.home"), ".aws"), + "credentials").getPath + if (sys.env.get("AWS_SHARED_CREDENTIALS_FILE").exists(_.trim.nonEmpty)) { + assert( + options(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY) == + sys.env("AWS_SHARED_CREDENTIALS_FILE")) + } else { + assert(options(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY) == expected) + } + val gsOptions = + NativeConfig.extractObjectStoreOptions(hadoopConf, new URI("gs://test-bucket/object")) + assert(!gsOptions.contains(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY)) + + // The env override wins, a blank override falls back to user.home, and HOME is not used. + assert( + NativeConfig.defaultSharedCredentialsFile( + Map("AWS_SHARED_CREDENTIALS_FILE" -> "/etc/aws/shared"), + "/synthetic/jvm-home") == "/etc/aws/shared") + assert( + NativeConfig.defaultSharedCredentialsFile( + Map("AWS_SHARED_CREDENTIALS_FILE" -> " ", "HOME" -> "/synthetic/env-home"), + "/synthetic/jvm-home") == "/synthetic/jvm-home/.aws/credentials") + } + test("extractObjectStoreOptions - multiple cloud provider configurations") { val hadoopConf = new Configuration() // S3A configs From f3758e59fa6a07a454770284e2222e08f7946478 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Wed, 16 Sep 2026 08:11:47 +0700 Subject: [PATCH 5/5] fix: resolve Hadoop's default credentials file on the executor at plan creation --- native/core/src/execution/jni_api.rs | 9 +++ native/core/src/execution/planner.rs | 7 +- native/core/src/parquet/objectstore/s3.rs | 76 +++++++++++++++++++ .../org/apache/comet/CometExecIterator.scala | 7 ++ .../comet/objectstore/NativeConfig.scala | 12 +-- .../apache/comet/exec/CometExecSuite.scala | 8 ++ .../comet/objectstore/NativeConfigSuite.scala | 17 +---- 7 files changed, 111 insertions(+), 25 deletions(-) diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs index 692b0d1bccf..50a78e79dec 100644 --- a/native/core/src/execution/jni_api.rs +++ b/native/core/src/execution/jni_api.rs @@ -116,6 +116,9 @@ use crate::execution::spark_config::{ COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES, }; use crate::parquet::encryption_support::{CometEncryptionFactory, ENCRYPTION_FACTORY_ID}; +use crate::parquet::objectstore::s3::{ + ExecutorObjectStoreDefaults, COMET_DEFAULT_PROFILE_FILE_KEY, +}; use datafusion_comet_proto::spark_operator::operator::OpStruct; use log::{info, warn}; use std::sync::OnceLock; @@ -744,6 +747,12 @@ fn prepare_datafusion_session_context( session_config.set_str("datafusion.execution.parquet.reorder_filters", "true"); } + // The executor JVM resolves the credentials file Hadoop's profile provider would read on + // this executor; scans overlay it onto their object store options. + session_config = session_config.with_extension(Arc::new(ExecutorObjectStoreDefaults { + default_profile_file: spark_config.get(COMET_DEFAULT_PROFILE_FILE_KEY).cloned(), + })); + // Pass through DataFusion configs from Spark. // e.g: spark-shell --conf spark.comet.datafusion.sql_parser.parse_float_as_decimal=true // becomes datafusion.sql_parser.parse_float_as_decimal=true diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 60c206617e3..0e12b55e4a4 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -100,6 +100,7 @@ use iceberg::expr::Bind; use crate::execution::operators::ExecutionError::GeneralError; use crate::execution::shuffle::{CometPartitioning, CompressionCodec}; use crate::execution::spark_plan::SparkPlan; +use crate::parquet::objectstore::s3::apply_executor_object_store_defaults; use crate::parquet::objectstore::s3_blob_fs_support::normalize_object_store_url; use crate::parquet::parquet_support::prepare_object_store_with_configs; use datafusion::common::scalar::ScalarStructBuilder; @@ -1695,11 +1696,12 @@ impl PhysicalPlanner { .map(|f| f.file_path.clone()) .expect("partition should have files after empty check"); - let object_store_options: HashMap = common + let mut object_store_options: HashMap = common .object_store_options .iter() .map(|(k, v)| (k.clone(), v.clone())) .collect(); + apply_executor_object_store_defaults(&self.session_ctx, &mut object_store_options); let (object_store_url, _, object_store_backend) = prepare_object_store_with_configs( self.session_ctx.runtime_env(), @@ -1747,11 +1749,12 @@ impl PhysicalPlanner { convert_spark_types_to_arrow_schema(scan.partition_schema.as_slice()); let projection_vector: Vec = scan.projection_vector.iter().map(|i| *i as usize).collect(); - let object_store_options: HashMap = scan + let mut object_store_options: HashMap = scan .object_store_options .iter() .map(|(k, v)| (k.clone(), v.clone())) .collect(); + apply_executor_object_store_defaults(&self.session_ctx, &mut object_store_options); let one_file = scan .file_partitions .first() diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index 26b3bd27f89..361d9baad0e 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -353,6 +353,43 @@ fn normalize_endpoint( }) } +/// Object store defaults the executor JVM resolves at plan creation and native applies to +/// every scan's options, so the native side reads the same files Hadoop does on that executor. +#[derive(Debug, Clone, Default)] +pub struct ExecutorObjectStoreDefaults { + /// The credentials file Hadoop's profile provider reads with no `fs.s3a.auth.profile.file` + /// configured, resolved against the executor JVM's `user.home` and environment. + pub default_profile_file: Option, +} + +/// The key the executor JVM and the native provider share for that file. +pub const COMET_DEFAULT_PROFILE_FILE_KEY: &str = "fs.s3a.comet.default.profile.file"; + +impl ExecutorObjectStoreDefaults { + /// Overlays the executor's defaults onto a scan's forwarded options. The executor value + /// wins over anything the driver serialized, since only the executor knows its own home. + pub fn apply(&self, options: &mut HashMap) { + if let Some(file) = &self.default_profile_file { + options.insert(COMET_DEFAULT_PROFILE_FILE_KEY.to_string(), file.clone()); + } + } +} + +/// Applies the executor defaults registered on `session` (see `ExecutorObjectStoreDefaults`) +/// to a scan's forwarded object store options. +pub fn apply_executor_object_store_defaults( + session: &datafusion::prelude::SessionContext, + options: &mut HashMap, +) { + if let Some(defaults) = session + .state() + .config() + .get_extension::() + { + defaults.apply(options); + } +} + /// The credentials file Hadoop's profile provider reads when none is configured: /// `AWS_SHARED_CREDENTIALS_FILE` when set, otherwise `~/.aws/credentials`. fn default_shared_credentials_file(env_override: Option, home: Option) -> String { @@ -2458,6 +2495,45 @@ mod tests { ); } + #[test] + fn test_executor_defaults_overlay_the_forwarded_options() { + use datafusion::prelude::{SessionConfig, SessionContext}; + let mut options = HashMap::from([( + COMET_DEFAULT_PROFILE_FILE_KEY.to_string(), + "/synthetic/driver-home/.aws/credentials".to_string(), + )]); + // Without executor defaults registered, the options pass through untouched. + apply_executor_object_store_defaults(&SessionContext::new(), &mut options); + assert_eq!( + options + .get(COMET_DEFAULT_PROFILE_FILE_KEY) + .map(String::as_str), + Some("/synthetic/driver-home/.aws/credentials") + ); + // The executor's own resolution wins over whatever the driver serialized. + let config = SessionConfig::new().with_extension(Arc::new(ExecutorObjectStoreDefaults { + default_profile_file: Some("/synthetic/executor-home/.aws/credentials".to_string()), + })); + apply_executor_object_store_defaults( + &SessionContext::new_with_config(config), + &mut options, + ); + assert_eq!( + options + .get(COMET_DEFAULT_PROFILE_FILE_KEY) + .map(String::as_str), + Some("/synthetic/executor-home/.aws/credentials") + ); + // An executor with nothing resolved leaves the options alone. + let config = + SessionConfig::new().with_extension(Arc::new(ExecutorObjectStoreDefaults::default())); + apply_executor_object_store_defaults( + &SessionContext::new_with_config(config), + &mut options, + ); + assert_eq!(options.len(), 1); + } + #[test] fn test_default_shared_credentials_file_matches_hadoop() { assert_eq!( diff --git a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala index e2c132904d5..37127bf5d1f 100644 --- a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala +++ b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala @@ -35,6 +35,7 @@ import org.apache.spark.util.SerializableConfiguration import org.apache.comet.CometConf._ import org.apache.comet.Tracing.withTrace import org.apache.comet.exceptions.CometQueryExecutionException +import org.apache.comet.objectstore.NativeConfig import org.apache.comet.parquet.CometFileKeyUnwrapper import org.apache.comet.serde.Config.ConfigMap import org.apache.comet.shuffle.ShufflePartitionPusher @@ -358,6 +359,12 @@ object CometExecIterator extends Logging { CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.key, CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED.get(SQLConf.get).toString) + // This runs on the executor, so the credentials file Hadoop's profile provider would read + // here is resolved against this JVM's user.home, not the driver's or the native process's. + builder.putEntries( + NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY, + NativeConfig.defaultSharedCredentialsFile()) + builder.build().toByteArray } diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index 28cac3e478c..0dee6c67f35 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -199,11 +199,12 @@ object NativeConfig { /** * Where the native side finds the credentials file Hadoop's profile provider would read with no * `fs.s3a.auth.profile.file` configured: `AWS_SHARED_CREDENTIALS_FILE`, else the JVM user's - * `~/.aws/credentials` (Hadoop resolves the home through `user.home`, not `HOME`). + * `~/.aws/credentials` (Hadoop resolves the home through `user.home`, not `HOME`). Resolved on + * the executor at plan creation, since the executor's home is what its Hadoop provider reads. */ val COMET_DEFAULT_PROFILE_FILE_KEY = "fs.s3a.comet.default.profile.file" - private[objectstore] def defaultSharedCredentialsFile( + private[comet] def defaultSharedCredentialsFile( env: Map[String, String] = sys.env, userHome: String = System.getProperty("user.home")): String = env @@ -246,13 +247,6 @@ object NativeConfig { val vendorPrefix = if (s3CompliantSchemes.contains(scheme)) s"fs.$scheme." else "" val vendorEntries = scala.collection.mutable.ArrayBuffer[(String, String)]() - // Hadoop's ProfileAWSCredentialsProvider reads this file when fs.s3a.auth.profile.file is - // unset; native resolves paths against its own environment, so the JVM's answer rides along - // for every scheme that uses the fs.s3a.* surface. - if (prefixes.get.contains("fs.s3a.")) { - options(COMET_DEFAULT_PROFILE_FILE_KEY) = defaultSharedCredentialsFile() - } - hadoopConf.iterator().asScala.foreach { entry => val key = entry.getKey if (prefixes.get.exists(prefix => key.startsWith(prefix))) { diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala index 5b5d43dbe8a..b5780bfd542 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala @@ -67,6 +67,14 @@ class CometExecSuite extends CometTestBase { } } + test("serialized executor configs carry the executor-resolved default credentials file") { + val protobuf = CometExecIterator.serializeCometSQLConfs() + val entries = org.apache.comet.serde.Config.ConfigMap.parseFrom(protobuf).getEntriesMap + assert( + entries.get(org.apache.comet.objectstore.NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY) == + org.apache.comet.objectstore.NativeConfig.defaultSharedCredentialsFile()) + } + test("SQLConf serde") { def roundtrip = { diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index 239d0a56b6a..1db398bd4e9 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -30,23 +30,12 @@ import org.apache.comet.CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY class NativeConfigSuite extends AnyFunSuite with Matchers { - test("extractObjectStoreOptions forwards the JVM-resolved default credentials file") { + test("the default credentials file is resolved like Hadoop and not frozen at planning") { + // Planning on the driver forwards nothing for it; the executor resolves it at plan creation. val hadoopConf = new Configuration() val options = NativeConfig.extractObjectStoreOptions(hadoopConf, new URI("s3a://test-bucket/object")) - val expected = new java.io.File( - new java.io.File(System.getProperty("user.home"), ".aws"), - "credentials").getPath - if (sys.env.get("AWS_SHARED_CREDENTIALS_FILE").exists(_.trim.nonEmpty)) { - assert( - options(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY) == - sys.env("AWS_SHARED_CREDENTIALS_FILE")) - } else { - assert(options(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY) == expected) - } - val gsOptions = - NativeConfig.extractObjectStoreOptions(hadoopConf, new URI("gs://test-bucket/object")) - assert(!gsOptions.contains(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY)) + assert(!options.contains(NativeConfig.COMET_DEFAULT_PROFILE_FILE_KEY)) // The env override wins, a blank override falls back to user.home, and HOME is not used. assert(