fix: honor the S3 profile name and file and Hadoop's addressing mode for custom endpoints

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`<br/>`software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider`                                                     | Access S3 using EC2 instance metadata service (IMDS)                                                            | None                                                                                                                            |
 | `com.amazonaws.auth.ContainerCredentialsProvider`<br/>`software.amazon.awssdk.auth.credentials.ContainerCredentialsProvider`<br/>`com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper` | Access S3 using ECS task credentials                                                                            | None                                                                                                                            |
 | `com.amazonaws.auth.WebIdentityTokenCredentialsProvider`<br/>`software.amazon.awssdk.auth.credentials.WebIdentityTokenFileCredentialsProvider`                                               | Authenticate using web identity token file                                                                      | None                                                                                                                            |
-| `com.amazonaws.auth.profile.ProfileCredentialsProvider`<br/>`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`<br/>`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://<bucket>.<endpoint>`, while path-style
+(`true`) sends them to `https://<endpoint>/<bucket>`, 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 `<bucket>.<host>`
+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

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"] }

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

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

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<String, String> = common
+                let mut object_store_options: HashMap<String, String> = 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<usize> =
                     scan.projection_vector.iter().map(|i| *i as usize).collect();
-                let object_store_options: HashMap<String, String> = scan
+                let mut object_store_options: HashMap<String, String> = 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()