sunchao commented on code in PR #6023: URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4050918438
########## 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: ### Correctness [P2] Select the delegate from the Spark/Hadoop profile Could this choose the provider from the active Spark/Hadoop profile instead of whether SDK v2 is present? The existing unconditional SDK v2 dependencies in `spark/pom.xml` put `DefaultCredentialsProvider` on the Spark 3.x test classpath too. This branch therefore selects the v2 class with Hadoop 3.3.4, whose provider factory requires `com.amazonaws.auth.AWSCredentialsProvider` and rejects it with `does not implement AWSCredentialsProvider`. I verified the two interfaces with class loading that did not initialize the default provider. The suite's initial Spark write cannot reach the native adapter regression on Spark 3.4/3.5. It would be worth exercising the selection with both SDKs present. ########## spark/src/main/spark-4.x/org/apache/comet/cloud/s3/AwsSdkCredentialProviderAdapter.java: ########## @@ -0,0 +1,110 @@ +/* + * 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.lang.reflect.Method; +import java.util.Map; + +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; + +import org.apache.comet.annotation.Public; +import org.apache.comet.util.ClassLoaders; + +/** + * Wraps a raw AWS SDK v2 {@link AwsCredentialsProvider} named via + * {@code fs.s3a.comet.credential.adapter.class}, for a provider not registered through S3A. This is + * the spark-4.x (SDK v2) body. Prefer {@link HadoopS3ACredentialProviderAdapter} unless the + * provider is a plain SDK class not wired through Hadoop. + * + * <pre> + * spark.hadoop.fs.s3a.comet.credential.provider.class=org.apache.comet.cloud.s3.AwsSdkCredentialProviderAdapter + * spark.hadoop.fs.s3a.comet.credential.adapter.class=<FQCN of an AwsCredentialsProvider> + * </pre> + */ +@Public +public class AwsSdkCredentialProviderAdapter implements CometS3CredentialProvider { + + static final String DELEGATE_CLASS_PROPERTY = "comet.credential.adapter.class"; + + 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) { + delegate = instantiate(bucket); + } + return delegate; + } + } + + private AwsCredentialsProvider instantiate(String bucket) throws Exception { + String className = AdapterSupport.lookup(properties, bucket, DELEGATE_CLASS_PROPERTY); + if (className == null) { + throw new IllegalStateException( + "AwsSdkCredentialProviderAdapter requires fs.s3a." + + DELEGATE_CLASS_PROPERTY + + " (or the per-bucket variant) to name an AwsCredentialsProvider"); + } + Class<?> clazz = ClassLoaders.loadClass(className); + if (!AwsCredentialsProvider.class.isAssignableFrom(clazz)) { + throw new IllegalStateException( + className + + " does not implement software.amazon.awssdk.auth.credentials.AwsCredentialsProvider"); + } + // SDK v2 instantiation conventions, in order: static create(), static builder().build(), + // public no-arg constructor. + Method create = AdapterSupport.staticMethod(clazz, "create"); + if (create != null) { + return (AwsCredentialsProvider) create.invoke(null); + } + Method builder = AdapterSupport.staticMethod(clazz, "builder"); + if (builder != null) { + Object b = builder.invoke(null); + Method build = b.getClass().getMethod("build"); + return (AwsCredentialsProvider) build.invoke(b); Review Comment: ### Correctness [P2] Invoke `build()` through an accessible builder API Could this call use the public builder interface or declared return type instead of the builder object's implementation class? A provider with a public `builder()` returning a public interface and a private implementation is valid, but `b.getClass().getMethod("build")` returns a method declared on that private class. I reproduced `IllegalAccessException` at `build.invoke(b)` with synthetic credentials, while calling the same public builder interface directly succeeds. This prevents the advertised builder-only provider path from constructing its delegate. It would be worth adding a fixture with a private builder implementation. ########## 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: ### Correctness [P2] Preserve S3A's credential-store path preparation Could both Hadoop adapter implementations prepare the credential-store path as `S3AFileSystem.initialize` does after applying bucket options? Hadoop 3.3.4 and 3.4.1 promote `fs.s3a.security.credential.provider.path` into `hadoop.security.credential.provider.path` before building the AWS provider list. The factory methods called here do not perform that step. When only this S3A path is configured, the forwarded map retains that key, but a provider looking up a secret through Hadoop's credential-provider API cannot see the configured store. This also affects per-bucket store paths and providers using stored secrets to obtain temporary credentials. An offline config-only delegate reproduced the missing generic path. Please cover this preparation with a synthetic credential-store test without forwarding raw secrets. ########## 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: ### Correctness [P2] Register the new suite with the CI suite checker Could this suite be added to the appropriate Linux and macOS workflow groups, or explicitly registered as a manual suite if that is the intended execution policy? The current Preflight run checks out merge `c7031bb` and exits 255 because `org.apache.comet.cloud.s3.HadoopS3ACredentialProviderAdapterBridgeSuite` is absent from `.github/workflows/pr_build_linux.yml`. `dev/ci/check-suites.py` requires each suite in both workflows unless it is explicitly exempted. This currently fails Required Checks and skips all runtime jobs. The [Preflight log](https://github.com/apache/datafusion-comet/actions/runs/35391902742/job/105751942281) confirms the failure. -- 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]
