parthchandra commented on code in PR #6023: URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4051264312
########## spark/src/test/scala/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapterBridgeSuite.scala: ########## @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.cloud.s3 + +import scala.collection.mutable +import scala.util.Try + +import org.apache.spark.SparkConf +import org.apache.spark.sql.SaveMode +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.functions.{col, sum} + +import org.apache.comet.CometS3TestBase + +/** + * End-to-end MinIO test for [[HadoopS3ACredentialProviderAdapter]] on the native Parquet path. + * + * The delegate is the AWS default credential chain -- a provider class Comet's native Rust list + * deliberately does NOT recognize. Without the adapter, the native reader fails with `Unsupported + * credential provider`; a successful read here proves the adapter routed credential resolution + * through Hadoop S3A instead. This is the regression from the spec's failure report. + * + * Credentials are supplied via JVM system properties (the AWS default chain reads them) rather + * than `fs.s3a.access.key` / `secret.key`, because Comet does not forward those secrets to the + * SPI. + */ +class HadoopS3ACredentialProviderAdapterBridgeSuite + extends CometS3TestBase + with AdaptiveSparkPlanHelper { Review Comment: Fixed. This is a MinIO/Docker suite, (like `CometS3CredentialBridgeSuite`, `ParquetReadFromS3Suite`, and `IcebergReadFromS3Suite`) so added to ignore list. ########## spark/src/main/spark-4.x/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapter.java: ########## @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.cloud.s3; + +import java.net.URI; +import java.util.Map; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.s3a.S3AUtils; +import org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory; +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; + +import org.apache.comet.annotation.Public; + +/** + * Delegates credential resolution to Hadoop S3A's own provider construction, so it accepts + * everything the {@code fs.s3a.aws.credentials.provider} chain accepts. This is the spark-4.x (AWS + * SDK v2) body; it calls {@link CredentialProviderListFactory} and returns v2 credentials. + * + * <p>Enable it (leaving {@code fs.s3a.aws.credentials.provider} untouched) with: + * + * <pre> + * spark.hadoop.fs.s3a.comet.credential.provider.class=org.apache.comet.cloud.s3.HadoopS3ACredentialProviderAdapter + * </pre> + */ +@Public +public class HadoopS3ACredentialProviderAdapter implements CometS3CredentialProvider { + + private Map<String, String> properties; + private volatile AwsCredentialsProvider delegate; + + @Override + public void initialize(Map<String, String> catalogProperties) { + this.properties = catalogProperties; + } + + @Override + public CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) + throws Exception { + AwsCredentialsProvider provider = ensureDelegate(context.getBucket()); + return SdkCredentialExtraction.toCometCredentials(provider.resolveCredentials()); + } + + private AwsCredentialsProvider ensureDelegate(String bucket) throws Exception { + AwsCredentialsProvider local = delegate; + if (local != null) { + return local; + } + synchronized (this) { + if (delegate == null) { + Configuration conf = + S3AUtils.propagateBucketOptions(AdapterSupport.toConfiguration(properties), bucket); + URI uri = new URI("s3a://" + bucket + "/"); + delegate = CredentialProviderListFactory.createAWSCredentialProviderList(uri, conf); Review Comment: Fixed. You're right that the factory methods don't do the credential-store promotion that `S3AFileSystem.initialize` does. I added a shared helper and call it in both Hadoop adapters right after `propagateBucketOptions`: AdapterSupport.patchSecurityCredentialProviders(conf); It copies `fs.s3a.security.credential.provider.path` into the generic `hadoop.security.credential.provider.path` (S3A path taking precedence), which is what lets a provider find secrets through Hadoop's credential-provider API. Since it runs after `propagateBucketOptions`, per-bucket store paths are already promoted to the base key. New test `resolvesKeysFromCredentialStoreViaS3aProviderPath` (v1 and v2): it writes the keys into a jceks store, forwards only the store path (no raw secrets), and asserts `SimpleAWSCredentialsProvider` resolves them — which only works with the patch in place. ########## spark/src/test/scala/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapterBridgeSuite.scala: ########## @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet.cloud.s3 + +import scala.collection.mutable +import scala.util.Try + +import org.apache.spark.SparkConf +import org.apache.spark.sql.SaveMode +import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper +import org.apache.spark.sql.functions.{col, sum} + +import org.apache.comet.CometS3TestBase + +/** + * End-to-end MinIO test for [[HadoopS3ACredentialProviderAdapter]] on the native Parquet path. + * + * The delegate is the AWS default credential chain -- a provider class Comet's native Rust list + * deliberately does NOT recognize. Without the adapter, the native reader fails with `Unsupported + * credential provider`; a successful read here proves the adapter routed credential resolution + * through Hadoop S3A instead. This is the regression from the spec's failure report. + * + * Credentials are supplied via JVM system properties (the AWS default chain reads them) rather + * than `fs.s3a.access.key` / `secret.key`, because Comet does not forward those secrets to the + * SPI. + */ +class HadoopS3ACredentialProviderAdapterBridgeSuite + extends CometS3TestBase + with AdaptiveSparkPlanHelper { + + override protected val testBucketName = "hadoop-adapter-bucket" + + // The AWS default-chain FQCN for whichever SDK the active Spark/Hadoop line ships (v2 on Spark + // 4.x, v1 on 3.x). Neither is in Comet's native provider list. + private val defaultChainClass: String = + if (Try( + Class.forName( + "software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider")).isSuccess) { + "software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider" + } else { + "com.amazonaws.auth.DefaultAWSCredentialsProviderChain" Review Comment: Fixed. The old check keyed off whether the v2 SDK class was on the classpath, but the v2 SDK is present on the Spark 3.x test classpath too (via Iceberg's S3 test deps), so it wrongly picked the v2 provider against Hadoop 3.3.4. The selection now keys off the Hadoop factory instead, which actually tracks the profile: Class.forName("org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory") That class only exists in Hadoop 3.4+ (the v2 line), so its presence reliably distinguishes the profiles. Spark 3.4/3.5 now select `com.amazonaws.auth.DefaultAWSCredentialsProviderChain` (the v1 class the 3.3.4 factory accepts), and the initial write can reach the native adapter regression. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
