diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index 97d3470facb..7d7a21a966e 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -221,7 +221,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 | None | +| `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}`. @@ -238,6 +239,16 @@ 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, +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. + ### 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 fabb56e2ec8..ef69ece0f20 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1971,6 +1971,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 bde2ba432c0..a3f8e6677b7 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -60,6 +60,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 f0c7735a503..c3ed9581461 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/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 c10fc60816e..361d9baad0e 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,33 @@ 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(), - ); - } - - // Extract endpoint configuration and modify if virtual hosted style is enabled - 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); + // 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 shape it for the selected addressing style. The flag is + // 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, + 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 +280,27 @@ 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 { + 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,17 +318,86 @@ fn normalize_endpoint( endpoint.to_string() }; - if virtual_hosted_style_request { - if endpoint.ends_with("/") { - Some(format!("{endpoint}{bucket}")) - } else { - Some(format!("{endpoint}/{bucket}")) + let path_style = |endpoint: String| { + Some(NormalizedEndpoint { + endpoint, + virtual_hosted_style_request: false, + }) + }; + 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 + 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, + }) +} + +/// 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()); } - } else { - Some(endpoint) // Avoid extra to_string() call since endpoint is already a String } } +/// 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 { + 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, @@ -323,6 +418,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"; @@ -358,6 +464,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"; @@ -481,7 +588,23 @@ 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. + // 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") + .or_else(|| get_non_empty_config(configs, bucket, "comet.default.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", source: format!("Unsupported credential provider: {credential_provider_name}").into(), @@ -712,7 +835,13 @@ enum CredentialProviderMetadata { Imds, Environment, WebIdentity, - Profile, + 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, access_key: String, @@ -735,7 +864,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 +880,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 +953,35 @@ 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, + credentials_only, + } => { + let mut builder = ProfileFileCredentialsProvider::builder() + .configure(&ProviderConfig::with_default_region().await); + if let Some(name) = name { + builder = builder.profile_name(name); + } + // 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) + .build(), + ); + } + Ok(Arc::new(builder.build())) } CredentialProviderMetadata::Static { is_valid, @@ -974,6 +1137,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 +1172,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 +1707,176 @@ 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 (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); + } + 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), + credentials_only: true, + }, + "provider {provider_name}, name {name:?}, file {file:?}" + ); + } + } + } + + #[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 result = + let test_provider = build_credential_provider(&configs, "test-bucket", Duration::from_secs(300)) .await - .unwrap(); - assert!(result.is_some(), "Should return a credential provider"); + .unwrap() + .expect("Should return a credential provider") + .metadata(); + assert_eq!( + test_provider, + CredentialProviderMetadata::Profile { + name: None, + file: None, + credentials_only: false, + }, + "provider {provider_name}" + ); + } + } - let test_provider = result.unwrap().metadata(); - assert_eq!(test_provider, CredentialProviderMetadata::Profile); + #[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(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") + .with_bucket_property( + "file-bucket", + "auth.profile.file", + "/etc/aws/bucket-credentials", + ) + .build(); + + 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()), + credentials_only: true, + }, + "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},{HADOOP_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()), + credentials_only: true, + }, + CredentialProviderMetadata::Imds, + ]) + ); + } + #[tokio::test] #[cfg_attr(miri, ignore)] // AWS credential providers call foreign functions async fn test_hadoop_iam_instance_credential_provider() { @@ -1966,75 +2325,509 @@ 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}" ); } + + // 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://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_custom_endpoint_with_virtual_hosted_style() { - let cases = vec![ - ( - "custom.endpoint.com", - "https://custom.endpoint.com/test-bucket", + 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()) + ); + + // 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] + 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!( + 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()) ), - ( - "https://custom.endpoint.com", - "https://custom.endpoint.com/test-bucket", + "/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 + // object_store appends the bucket itself + let cases = [ + ("custom.endpoint.com", "https://custom.endpoint.com"), + ("http://custom.endpoint.com", "http://custom.endpoint.com"), ( "https://custom.endpoint.com/", - "https://custom.endpoint.com/test-bucket", + "https://custom.endpoint.com/", ), + ("http://minio.internal:9000", "http://minio.internal:9000"), ( + "https://custom.endpoint.com:8443/", + "https://custom.endpoint.com:8443/", + ), + ( + "https://custom.endpoint.com/path/to/resource", "https://custom.endpoint.com/path/to/resource", - "https://custom.endpoint.com/path/to/resource/test-bucket", ), ( - "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"), + ( + 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 +2852,29 @@ 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, + 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(), + "Profile(name: analytics, file: /etc/aws/credentials)" + ); + // Test Chain provider let chain_metadata = CredentialProviderMetadata::Chain(vec![ CredentialProviderMetadata::Static { 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 75328649fea..0dee6c67f35 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,22 @@ 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`). 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[comet] 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") 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 8e5a60d63b0..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,6 +30,24 @@ import org.apache.comet.CometConf.COMET_S3_COMPLIANT_SCHEMES_KEY class NativeConfigSuite extends AnyFunSuite with Matchers { + 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")) + 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( + 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