andygrove commented on code in PR #6023: URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4073555584
########## spark/src/main/spark-4.x/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapter.java: ########## @@ -0,0 +1,97 @@ +/* + * 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 org.apache.comet.annotation.Public; + +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; + +/** + * 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); + AdapterSupport.patchSecurityCredentialProviders(conf); + URI uri = new URI("s3a://" + bucket + "/"); + try { + delegate = CredentialProviderListFactory.createAWSCredentialProviderList(uri, conf); Review Comment: One more place the adapter diverges from what `S3AFileSystem.initialize` actually does. At `S3AFileSystem.java:1117` the factory call is the else branch. When `hasDelegationTokenBinding(conf)` is true Spark takes `tokens.getCredentialProviders()` instead, and the comment right above it says "switch to the DT provider and bypass all other configured providers". This body always calls the factory, so on a cluster using S3A delegation tokens Comet resolves the configured chain while Spark resolves the DT, which is the silent identity swap again. `fs.s3a.delegation.token.binding` is an `fs.s3a.*` key so it is already in the forwarded map. Would checking for it here and failing with a message naming it be enough? Implementing DT support is clearly its own piece of work, I would just rather it not resolve the wrong principal in the meantime. Same for the spark-3.x body. ########## spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcher.java: ########## @@ -74,19 +79,43 @@ public static long ensureInitialized( catalogProperties == null ? Collections.emptyMap() : Collections.unmodifiableMap(new HashMap<>(catalogProperties)); + // Key on a digest of the properties, not the values themselves: the KEY_TO_HANDLE map is + // static and lives for the JVM lifetime, and the property bag may carry secrets (vended + // credentials, static keys). The full map is still handed to initialize() below; only the + // long-lived cache key is reduced to a digest. A distinct config still yields a distinct key. InstanceKey key = - new InstanceKey(providerClassName, dispatchKey == null ? "" : dispatchKey, snapshot); + new InstanceKey( + providerClassName, dispatchKey == null ? "" : dispatchKey, digestOf(snapshot)); return KEY_TO_HANDLE.computeIfAbsent( key, k -> { - CometS3CredentialProvider provider = instantiate(k.providerClassName); - provider.initialize(k.catalogProperties); + CometS3CredentialProvider provider = instantiate(providerClassName); + provider.initialize(snapshot); long handle = HANDLE_SEQ.getAndIncrement(); INSTANCES.put(handle, new RegisteredProvider(provider, k)); return handle; }); } + /** Stable SHA-256 digest of the property bag, so secret values are not retained in the key. */ + private static String digestOf(Map<String, String> props) { Review Comment: Thanks for taking on the retention point rather than deferring it. Reading it back though, I do not think it gets where the comment says it does. `INSTANCES` is static too, and both adapters hold the map in `this.properties` for as long as the provider lives, so the same values sit in the JVM for the same lifetime whether or not the key is digested. Given that, and given the encoding ambiguity @sunchao found on this method, is the digest worth keeping? Reverting to the map key makes that finding go away for free. If you want to keep it, then clearing `properties` in both adapters once the delegate is built would be the piece that actually shortens the exposure, and the digest would need the length-prefixed encoding from the other thread. ########## docs/source/contributor-guide/s3-credential-provider-design.md: ########## @@ -93,6 +95,19 @@ The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The `IcebergScanExec` derives a redacting `Debug` so plan dumps and tracing do not leak the property bag. +## Property-bag handling on the Parquet path + +The Parquet path forwards the full `fs.s3a.*` config subset as `catalog_properties`, so an SPI provider sees exactly the config Spark would. `forward_catalog_properties` in `native/core/src/parquet/objectstore/s3.rs` keeps every `fs.s3a.*` key, including the static credentials (`*.access.key`, `*.secret.key`, `*.session.token`). Forwarding them is deliberate: `AWSCredentialProviderList` skips a provider that throws `NoAwsCredentialsException` and moves to the next entry, so stripping the static keys would let a chain such as `SimpleAWSCredentialsProvider,customProvider` silently resolve through a different provider than Spark, reading data as a different principal. These keys already cross JNI for the non-adapter native path (`build_credential_provider` reads them), and the Iceberg path already forwards its full bag, so this matches existing behavior rather than widening exposure. As on the Iceberg path, the dispatcher retains the forwarded map in its instance-cache key for the JVM lifetime. Review Comment: This paragraph ends with "the dispatcher retains the forwarded map in its instance-cache key for the JVM lifetime", which contradicts line 54 saying the key holds a SHA-256 digest. Looks like it survived from before the digest commit. Whichever way the digest question lands, could these two say the same thing? Two comments went stale with the same change. `spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapterTest.java:38` still says "the native path strips them from the forwarded map", and `HadoopS3ACredentialProviderAdapterBridgeSuite.scala:40` says credentials go through system properties "because Comet does not forward those secrets to the SPI". ########## spark/src/test/scala/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapterBridgeSuite.scala: ########## @@ -0,0 +1,108 @@ +/* + * 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 must match what the active Hadoop-aws line's provider factory + // accepts, not merely which SDK jar is on the test classpath: the v2 SDK is present on the + // Spark 3.x test classpath too (Iceberg's S3 test deps), but Hadoop 3.3.4's factory only accepts + // the v1 interface. CredentialProviderListFactory exists only in Hadoop 3.4+ (the v2 line), so + // its presence is the reliable per-profile signal. Neither class is in Comet's native list. + private val defaultChainClass: String = + if (Try( + Class.forName("org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory")).isSuccess) { + "software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider" + } else { + "com.amazonaws.auth.DefaultAWSCredentialsProviderChain" + } + + private val savedProps = mutable.Map[String, String]() + private def setProp(key: String, value: String): Unit = { + savedProps(key) = System.getProperty(key) + System.setProperty(key, value) + } + + override protected def sparkConf: SparkConf = { + val conf = super.sparkConf + conf.set( + "spark.hadoop.fs.s3a.comet.credential.provider.class", + classOf[HadoopS3ACredentialProviderAdapter].getName) + conf.set("spark.hadoop.fs.s3a.aws.credentials.provider", defaultChainClass) + } + + override def beforeAll(): Unit = { + // Both the v1 (aws.secretKey) and v2 (aws.secretAccessKey) secret property names are set so the + // default chain resolves regardless of which SDK is on the classpath. + setProp("aws.accessKeyId", userName) + setProp("aws.secretKey", password) + setProp("aws.secretAccessKey", password) + super.beforeAll() + } + + override def afterAll(): Unit = { + super.afterAll() + savedProps.foreach { + case (key, null) => System.clearProperty(key) + case (key, value) => System.setProperty(key, value) + } + } + + test( Review Comment: Thanks for running this on 4.0 and 4.1 and putting it in the description. Following on from the stale comment above, that system-property route means nothing anywhere exercises the forwarding end to end. The Rust test covers the native half, the Java tests hand-build the map, and this suite deliberately routes around `fs.s3a.access.key`. So the change this revision is built around, the full map crossing JNI and the adapter rebuilding a `Configuration` from it, has no test joining the two halves. Since the suite is manual anyway and costs nothing in CI, could you add a second case using `fs.s3a.access.key` and `fs.s3a.secret.key` with `fs.s3a.aws.credentials.provider` set to `org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider,<something the native list rejects>`? That is exactly the chain from my earlier comment, and it fails if the static keys ever stop being forwarded. ########## native/core/src/parquet/objectstore/s3.rs: ########## @@ -334,6 +337,22 @@ fn lookup_provider_class<'a>( get_config_trimmed(configs, bucket, PROVIDER_CLASS_PROPERTY).filter(|s| !s.is_empty()) } +/// Builds the `catalog_properties` map forwarded to the SPI on the Parquet path: the full +/// `fs.s3a.*` subset. This matches the Iceberg path, which forwards its full property bag, so an +/// adapter delegating to Hadoop's provider construction sees exactly the config Spark would -- +/// including the static keys a provider chain may resolve through. Stripping them would let a +/// chain like `SimpleAWSCredentialsProvider,customProvider` silently resolve through a different +/// entry than Spark, reading data as a different principal. These keys already cross JNI for the +/// non-adapter path (see `build_credential_provider`). HashMap equality is order-independent, so +/// the dispatcher instance-cache key stays stable regardless of iteration order. +fn forward_catalog_properties(configs: &HashMap<String, String>) -> HashMap<String, String> { Review Comment: Thanks for taking the full-forwarding route, that closes the chain fall-through cleanly. I think there is one more key on the far side of the same problem though. `NativeConfig.extractObjectStoreOptions` only collects `fs.s3a.*` plus the two `fs.comet.*` scheme keys, so by the time `configs` gets here the `fs.s3a.` filter cannot actually drop anything real. The bit that matters is what never arrives: `hadoop.security.credential.provider.path`. Hadoop's javadoc on `S3A_SECURITY_CREDENTIAL_PROVIDER_PATH` describes it as an "extra set of security credentials which will be prepended to that set in `hadoop.security.credential.provider.path`", so `patchSecurityCredentialProviders` is handling the extra spelling and the base one is the one most deployments set. That gives an inconsistency that is hard to debug. `new Configuration()` in `AdapterSupport.toConfiguration` loads the executor's `core-site.xml`, so it works if the jceks path is there, and it silently does not if the user set `spark.hadoop.hadoop.security.credential.provider.path`. When it does not, `SimpleAWSCredentialsProvider` resolves nothing, `AWSCredentialProviderList` moves on, and we are back to reading as a different principal. Could `AdapterSupport.toConfiguration` seed from the executor's own Hadoop conf and overlay the forwarded `fs.s3a.*` on top? That covers this key and anything else a provider reaches for that is not `fs.s3a.*`. If you would rather keep the round trip, adding the one key to what `NativeConfig` forwards works too. Related, the new Rust test asserts `spark.master` is filtered out, but that key can never be in `configs`. Could it assert on something that can be, so the test is exercising a real boundary? ########## spark/src/main/spark-4.x/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapter.java: ########## @@ -0,0 +1,97 @@ +/* + * 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 org.apache.comet.annotation.Public; + +import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; + +/** + * 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 { Review Comment: `ensureDelegate` takes a bucket, feeds it to `propagateBucketOptions` and the URI, then caches the result without the bucket in the key. That is correct today only because `create_store` passes `bucket` as the dispatch key, so the Parquet path gets a separate instance per bucket. `iceberg_common.rs:177` passes the catalog name when one is available, so a single instance there serves every bucket in the catalog and the second bucket gets the first bucket's delegate. Both classes are `@Public`, so nothing stops someone naming one as the Iceberg provider class, and the user guide advertises `fs.s3a.bucket.<bucket>.comet.credential.adapter.class` as a supported per-bucket knob. Could the cache be a small `ConcurrentHashMap<String, ...>` keyed by bucket? A test that resolves two buckets with different per-bucket keys through one adapter instance would pin it down. This applies to all four adapter bodies. -- 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]
