parthchandra commented on code in PR #6023:
URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4065824295
##########
native/core/src/parquet/objectstore/s3.rs:
##########
@@ -334,6 +337,23 @@ fn lookup_provider_class<'a>(
get_config_trimmed(configs, bucket, PROVIDER_CLASS_PROPERTY).filter(|s|
!s.is_empty())
}
+/// Suffixes of `fs.s3a.*` keys that carry static-credential secrets. These
are deliberately not
+/// forwarded to the SPI: the adapters exist for the case where static keys
are not used, and
+/// forwarding secrets would widen the blast radius and put them in the
dispatcher cache-key hash.
+const SECRET_KEY_SUFFIXES: [&str; 3] = [".access.key", ".secret.key",
".session.token"];
Review Comment:
Fixed. `forward_catalog_properties` now forwards the full `fs.s3a.*` subset
with nothing stripped,
so a chain like `SimpleAWSCredentialsProvider,custom` (and assumed-role)
resolves the same
principal as Spark instead of silently falling through to a different entry.
You were right that the blast-radius argument didn't hold: these keys
already cross JNI for the
non-adapter path (`build_credential_provider`), and the Iceberg path already
forwards its full bag.
I also addressed the retention point you raised rather than deferring it:
the dispatcher's
`InstanceKey` now stores a SHA-256 digest of the property bag instead of the
raw values, so the
static JVM-lifetime cache map holds no secrets. `initialize()` still gets
the full map, and a
changed secret value changes the digest so the instance is rebuilt
correctly. The dispatcher's
isolation/multi-tenant tests still pass.
##########
native/core/src/parquet/objectstore/s3.rs:
##########
@@ -1023,6 +1043,40 @@ mod tests {
assert_eq!(session_token, Some("test_session_token"));
}
+ #[test]
+ fn test_forward_catalog_properties_filters_scope_and_secrets() {
+ let mut configs: HashMap<String, String> = HashMap::new();
+ configs.insert(
+ "fs.s3a.aws.credentials.provider".to_string(),
+
"com.amazonaws.auth.DefaultAWSCredentialsProviderChain".to_string(),
+ );
+ configs.insert("fs.s3a.endpoint".to_string(),
"s3.example.com".to_string());
+ configs.insert(
+ format!("fs.s3a.comet.{PROVIDER_CLASS_PROPERTY}"),
Review Comment:
Fixed — was building `fs.s3a.comet.comet.credential.provider.class`. Now
`format!("fs.s3a.{PROVIDER_CLASS_PROPERTY}")`, and the test asserts the
activation key survives
forwarding.
##########
native/core/src/parquet/objectstore/s3.rs:
##########
@@ -334,6 +337,23 @@ fn lookup_provider_class<'a>(
get_config_trimmed(configs, bucket, PROVIDER_CLASS_PROPERTY).filter(|s|
!s.is_empty())
}
+/// Suffixes of `fs.s3a.*` keys that carry static-credential secrets. These
are deliberately not
+/// forwarded to the SPI: the adapters exist for the case where static keys
are not used, and
+/// forwarding secrets would widen the blast radius and put them in the
dispatcher cache-key hash.
+const SECRET_KEY_SUFFIXES: [&str; 3] = [".access.key", ".secret.key",
".session.token"];
+
+/// Builds the `catalog_properties` map forwarded to the SPI on the Parquet
path: the `fs.s3a.*`
+/// subset with static-credential secrets removed. 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> {
+ configs
+ .iter()
+ .filter(|(k, _)| k.starts_with("fs.s3a."))
+ .filter(|(k, _)| !SECRET_KEY_SUFFIXES.iter().any(|suffix|
k.ends_with(suffix)))
Review Comment:
`fs.s3a.encryption.key`,`proxy.password`, etc. are now forwarded the same
as everything else, exactly as Spark sees them. And the dispatcher no longer
retains any of these values
##########
dev/ci/check-suites.py:
##########
@@ -37,6 +37,7 @@ def file_to_class_name(path: Path) -> str | None:
"org.apache.comet.parquet.ParquetReadFromFakeHadoopFsSuite", # manual
test suite (loads libhdfs, see #5023)
"org.apache.comet.IcebergReadFromS3Suite", # manual test suite
"org.apache.comet.cloud.s3.CometS3CredentialBridgeSuite", # manual
test suite
+
"org.apache.comet.cloud.s3.HadoopS3ACredentialProviderAdapterBridgeSuite", #
manual test suite
Review Comment:
Ran the suite locally on the spark-4.0, and spark-4.1 profile against a
MinIO testcontainer. Updated the PR description
##########
spark/src/main/spark-3.x/org/apache/comet/cloud/s3/SdkCredentialExtraction.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * 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 com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSSessionCredentials;
+
+/**
+ * Maps AWS SDK v1 {@link AWSCredentials} onto {@link CometS3Credentials}.
Compiled only into
+ * spark-3.4 / 3.5 builds (spark-3.x source set), directly against SDK v1.
+ */
+final class SdkCredentialExtraction {
+
+ private SdkCredentialExtraction() {}
+
+ static CometS3Credentials toCometCredentials(AWSCredentials creds) {
+ String sessionToken = null;
+ if (creds instanceof AWSSessionCredentials) {
+ sessionToken = ((AWSSessionCredentials) creds).getSessionToken();
+ }
+ // The v1 base interface exposes no expiration; report 0 (unknown). Safe:
the Parquet path
+ // ignores expiration and the Iceberg path applies a bounded default TTL.
+ return new CometS3Credentials(creds.getAWSAccessKeyId(),
creds.getAWSSecretKey(), sessionToken, 0L);
Review Comment:
Fixed. Added `src/main/spark-*/**/*.java` and `src/test/spark-*/**/*.java`
to the `<java>` spotless
block, parallel to the existing `<scala>` includes. `make format` now
reaches the adapter code (it
wrapped the two >100-col lines) and alsoupdated a few pre-existing shim
files.
##########
spark/src/main/spark-4.x/org/apache/comet/cloud/s3/HadoopS3ACredentialProviderAdapter.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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);
+ AdapterSupport.patchSecurityCredentialProviders(conf);
+ URI uri = new URI("s3a://" + bucket + "/");
+ delegate =
CredentialProviderListFactory.createAWSCredentialProviderList(uri, conf);
Review Comment:
Went with your fallback: both Hadoop adapters now catch `LinkageError`
around the factory call and
rethrow naming the Hadoop/SDK line the build targets, so an EMR-style
mismatch fails legibly
instead of a bare `NoSuchMethodError`. A true runtime probe would need both
SDK bodies compiled in,
which fights the per-source-set design, so I'd keep that out of this PR
unless you feel strongly.
##########
spark/src/main/spark-4.x/org/apache/comet/cloud/s3/AwsSdkCredentialProviderAdapter.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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");
Review Comment:
Fixed both. There's now one shared `AdapterSupport.instantiateDelegate` used
by v1 and v2, so the
`@Public` contract is identical: `(URI, Configuration)` -> `(Configuration)`
-> `create()` ->
`builder().build()` -> `getInstance()` -> no-arg. A Hadoop-convention
provider now works the same
on Spark 3.5 and 4.0.
The factory lookup also requires the return type to be assignable to the
target provider (matching
Hadoop's `getFactoryMethod`), so an unrelated `static String
create()`/`getInstance()` is skipped
rather than invoked and failing at the cast. Added a
`V2UnrelatedCreateProvider` fixture and test,
plus the private-builder-impl test for the `build()` case.
##########
docs/source/user-guide/latest/s3-credential-providers.md:
##########
@@ -36,6 +36,39 @@ You probably do, if any of these are true:
- You have a custom Iceberg `client.factory` that injects a configured S3
client.
- Spark queries against your S3 paths work, but the same queries with Comet
enabled fail with 403.
+## Built-in adapters
+
+If a native Parquet scan fails with `Unsupported credential provider: <class>`
(for example `com.amazonaws.auth.DefaultAWSCredentialsProviderChain`), the
class you named in `fs.s3a.aws.credentials.provider` is one that plain
Spark/Hadoop accepts but Comet's native reader does not reimplement. Comet
ships two built-in `CometS3CredentialProvider` adapters that fix this with a
one-line config change; you leave your existing
`fs.s3a.aws.credentials.provider` untouched.
+
+These adapters cover the Parquet native scan path only. Enabling one is
opt-in: naming it is what activates it, and Comet's existing native provider
handling is unchanged for everyone else.
+
+### `HadoopS3ACredentialProviderAdapter` (recommended)
+
+Delegates to Hadoop S3A's own provider construction, so it accepts everything
the `fs.s3a.aws.credentials.provider` chain accepts (the default chain,
web-identity, assumed-role, custom signers, per-bucket config). This is the
general answer for the failure above.
Review Comment:
Updated. -- the section now just says the chain (static keys, assumed-role,
etc.) resolves the same way it would under Spark.
--
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]
