parthchandra commented on code in PR #6023:
URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4077638324
##########
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) {
+ MessageDigest md;
+ try {
+ md = MessageDigest.getInstance("SHA-256");
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 not available", e);
+ }
+ // Sort by key for order-independence; NUL separators avoid key/value
boundary ambiguity.
+ for (Map.Entry<String, String> e : new TreeMap<>(props).entrySet()) {
+ md.update(e.getKey().getBytes(StandardCharsets.UTF_8));
+ md.update((byte) 0);
+ String v = e.getValue();
+ md.update(v == null ? new byte[] {1} :
v.getBytes(StandardCharsets.UTF_8));
+ md.update((byte) 0);
Review Comment:
Resolved by removing the digest entirely rather than fixing its encoding.
Per the discussion with
Andy on this method, the digest wasn't actually reducing secret retention
(INSTANCES and the
provider hold the map for the same lifetime), so I reverted InstanceKey to
the exact map-based key.
That makes the U+0000 input-encoding ambiguity moot -- there's no
serialization step anymore. The dispatcher isolation/multi-tenant tests still
pass.
##########
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:
Went with (a). AdapterSupport.toConfiguration now seeds from the executor's
own Spark-derived Hadoop conf (spark.hadoop.* via SparkEnv) and overlays the
forwarded fs.s3a.* on top, so hadoop.security.credential.provider.path -- and
anything else a provider reads that isn't
fs.s3a.* -- is present. Off the executor (unit tests) it falls back to a
bare Configuration.
Also fixed the Rust test to assert a realistic dropped key (fs.comet.*)
rather than spark.master.
##########
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:
All four adapters now cache one delegate per bucket (ConcurrentHashMap), so
a single instance on the Iceberg path serves each bucket correctly.
Added a unit test resolving two buckets with different per-bucket keys
through one instance
##########
spark/src/main/java/org/apache/comet/cloud/s3/AdapterSupport.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.Constructor;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.net.URI;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+
+/** Config and reflection helpers shared by the built-in S3 credential
provider adapters. */
+final class AdapterSupport {
+
+ private AdapterSupport() {}
+
+ /**
+ * Rebuilds a Hadoop {@link Configuration} from the forwarded {@code
fs.s3a.*} map. The adapters
+ * run on the executor without a live {@code S3AFileSystem}, so keys are
copied onto a fresh
+ * Configuration (which still loads core-site defaults).
+ */
+ static Configuration toConfiguration(Map<String, String> props) {
+ Configuration conf = new Configuration();
+ for (Map.Entry<String, String> entry : props.entrySet()) {
+ if (entry.getValue() != null) {
+ conf.set(entry.getKey(), entry.getValue());
+ }
+ }
+ return conf;
+ }
+
+ /**
+ * Per-bucket then global lookup, mirroring Comet's native {@code fs.s3a}
resolution: {@code
+ * fs.s3a.bucket.<bucket>.<property>} wins over {@code fs.s3a.<property>}.
Returns null if neither
+ * is set (after trimming).
+ */
+ static String lookup(Map<String, String> props, String bucket, String
property) {
+ String perBucket = props.get("fs.s3a.bucket." + bucket + "." + property);
+ if (perBucket != null && !perBucket.trim().isEmpty()) {
+ return perBucket.trim();
+ }
+ String global = props.get("fs.s3a." + property);
+ if (global != null && !global.trim().isEmpty()) {
+ return global.trim();
+ }
+ return null;
+ }
+
+ /** Returns the public static no-arg method {@code name} on {@code clazz},
or null if absent. */
+ private static Method staticMethod(Class<?> clazz, String name) {
+ try {
+ Method m = clazz.getMethod(name);
+ return Modifier.isStatic(m.getModifiers()) ? m : null;
+ } catch (NoSuchMethodException e) {
+ return null;
+ }
+ }
+
+ /**
+ * Instantiates a credential-provider delegate, trying the same ordered
conventions for both the
+ * v1 and v2 adapters so their {@code @Public} contract is identical: the
Hadoop-style {@code
+ * (URI, Configuration)} and {@code (Configuration)} constructors first
(matching {@code
+ * S3AUtils.getInstanceFromReflection}), then the SDK static factories
{@code create()} / {@code
+ * builder().build()} / {@code getInstance()}, then a public no-arg
constructor.
+ *
+ * <p>Factory return types must be assignable to {@code targetType} (as
Hadoop's {@code
+ * getFactoryMethod} requires), so an unrelated {@code static String
create()} is skipped rather
+ * than invoked and failing later with a {@code ClassCastException}. Returns
an untyped instance;
+ * the caller casts to its SDK provider interface.
+ */
+ static Object instantiateDelegate(
+ Class<?> targetType, Class<?> clazz, URI uri, Configuration conf) throws
Exception {
+ Constructor<?> uriConf = constructor(clazz, URI.class,
Configuration.class);
+ if (uriConf != null) {
+ return uriConf.newInstance(uri, conf);
+ }
+ Constructor<?> confOnly = constructor(clazz, Configuration.class);
+ if (confOnly != null) {
+ return confOnly.newInstance(conf);
+ }
+ Method create = factoryMethod(clazz, "create", targetType);
+ if (create != null) {
+ return create.invoke(null);
+ }
+ Method builder = staticMethod(clazz, "builder");
+ if (builder != null) {
+ // Resolve build() off builder()'s declared (public) return type, not
the runtime object's
+ // class, which may be a non-public implementation.
+ Method build = builder.getReturnType().getMethod("build");
+ if (targetType.isAssignableFrom(build.getReturnType())) {
Review Comment:
Fixed. The builder path now checks the built instance's runtime type instead
of build()'s declared return type, so a public Builder that inherits an erased
`Object build()` from a generic `SdkBuilder<B, T>` is accepted.
Added a V2GenericBuilderProvider fixture (build() inherited from a generic
super-interface, not redeclared) plus a test; the existing concrete-return
fixture stays as the covariant case.
##########
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:
Reverted to the map-based key as you suggested -- it's less code and makes
@sunchao's encoding finding moot. The retention is inherent to
INSTANCES/provider anyway, so the digest wasn't buying anything.
##########
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:
Fixed all of them -- the design doc keying/forwarding paragraphs no longer
mention a digest and now describe the executor-conf seeding; the v2 test and
bridge-suite comments no longer claim secrets are stripped.
##########
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:
Added a second MinIO case:
fs.s3a.aws.credentials.provider=SimpleAWSCredentialsProvider,<default chain>
with static keys and no system properties, so the fallback can't resolve and
the read succeeds only if the static keys crossed JNI and reached
SimpleAWSCredentialsProvider. Moved the default-chain test's system properties
into that test so this one is a real regression check.
##########
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:
Added a guard: both Hadoop adapters now fail with a message naming
fs.s3a.delegation.token.binding when it's set, rather than building the chain
and resolving a different principal than Spark's DT provider. DT support
remains its own piece of work.
--
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]
