andygrove commented on code in PR #6023:
URL: https://github.com/apache/datafusion-comet/pull/6023#discussion_r4063549929
##########
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:
I think this filter can silently change which identity Comet reads with,
which is the one thing I do not want the adapters to be able to do.
`AWSCredentialProviderList.resolveCredentials()` catches
`NoAwsCredentialsException`, logs at debug and continues to the next provider.
`SimpleAWSCredentialsProvider.resolveCredentials()` throws exactly that when
the keys are empty. So with the keys stripped, a chain like
`org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider,com.vendor.Custom` does
not error. It resolves through the vendor provider instead. Spark reads that
data as the static-key principal, Comet reads it as the vendor one, same
config, no warning.
That chain shape is not exotic, it is close to the reason someone reaches
for the adapter in the first place, since the native list rejects a chain if
any single entry is unrecognised.
Assumed-role has the same problem from a different direction.
`Constants.ASSUMED_ROLE_CREDENTIALS_DEFAULT` is
`SimpleAWSCredentialsProvider.NAME`, so the default inner provider reads the
static keys and the STS call has nothing to authenticate with.
On the blast-radius argument in the comment, `extractObjectStoreOptions`
already ships `access.key`, `secret.key` and `session.token` across JNI and
`build_credential_provider` reads them at lines 497-500, so they are in this
process on both sides already. Is the part you actually want to avoid the
dispatcher retaining them in `InstanceKey` for the JVM lifetime? If so, could
we hash a redacted view for the cache key and still pass the full map to
`initialize()`? The other way out would be having the adapter build its
`Configuration` from the executor's own Hadoop conf rather than round-tripping
through native, which sidesteps the question entirely.
If you would rather keep the stripping, could the Hadoop adapter at least
detect a static-credential provider in the resolved list with no keys present
and fail naming that, instead of falling through?
##########
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:
I think this key is doubled. `PROVIDER_CLASS_PROPERTY` is already
`comet.credential.provider.class` and `get_config` supplies the `fs.s3a.`
prefix, so this builds `fs.s3a.comet.comet.credential.provider.class`. Should
be `format!("fs.s3a.{PROVIDER_CLASS_PROPERTY}")`.
There is also no assertion on it further down, so the case it was presumably
meant to cover, that the activation key itself survives forwarding, is not
actually being tested.
##########
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:
Separate from the identity question above, I do not think these three
suffixes cover what the comment says they cover. Hadoop's own default
`hadoop.security.sensitive-config-keys` in `core-default.xml` also lists
`fs.s3a.encryption.key`, `fs.s3a.server-side-encryption.key`,
`fs.s3a.*.server-side-encryption.key` and `fs.s3a.session.key`, and its
`password$` pattern picks up `fs.s3a.proxy.password`. For SSE-C,
`fs.s3a.encryption.key` is the raw base64 AES-256 key.
Those all get handed to whatever third-party `CometS3CredentialProvider` the
user named and then held for the JVM lifetime in the dispatcher's static
`InstanceKey`, which is the outcome the comment above says we are avoiding.
Would anchoring the suffix list to Hadoop's own redaction list be reasonable?
##########
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:
The v1 and v2 bodies are picked by Spark profile, but what really decides
which one works is the hadoop-aws line on the cluster, and those two are not
tied together. EMR 7.x is Spark 3.5 on Hadoop 3.4.x, and there the spark-3.5
Comet build carries the v1 body while Hadoop 3.4's
`createAWSCredentialProviderSet` returns a v2 `AWSCredentialProviderList`.
Return type is part of the method descriptor, so that is a `NoSuchMethodError`
with nothing in it pointing at the cause. That is uncomfortably close to the
environment described in #6022.
The interesting bit is that `HadoopS3ACredentialProviderAdapterBridgeSuite`
already solves this. It probes for `CredentialProviderListFactory` and the
comment there says outright that the Spark profile is not the reliable signal
and class presence is. Could production use the same probe? If that is more
surgery than you want here, would catching `LinkageError` in `ensureDelegate`
and rethrowing with the Hadoop and SDK line we expected be enough to make the
failure legible?
##########
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:
This line is 104 columns, and the `does not implement` message in the
spark-4.x `AwsSdkCredentialProviderAdapter` is 102, both past
google-java-format's 100. Not your fault. This PR is the first to put Java
under `src/main/spark-*`, and spotless 2.43.0 builds its Java default includes
from `Build.getSourceDirectory()` and `getTestSourceDirectory()`, so it only
ever sees `src/main/java` and `src/test/java` and misses everything
build-helper adds. The `<scala>` block in the root pom already works around
this with explicit `src/main/spark-*/**/*.scala` includes, the `<java>` block
just never got the parallel ones.
Could you add them in this PR? Otherwise none of the new adapter code is
reachable by `make format` and it will drift.
##########
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:
Hadoop 3.4.1's `S3AUtils.getInstanceFromReflection` tries the `(URI,
Configuration)` and `(Configuration)` constructors before it reaches the
`create()` factory, which is the same ordering the v1 path uses. This body
skips both constructor forms, so a provider written to the Hadoop convention
works through the v1 adapter on Spark 3.5 and stops working on Spark 4.0 under
the same class name. Given both are `@Public` and pinned for 1.x, I would
rather get the contract identical now.
Related, `AdapterSupport.staticMethod` checks public and static but not the
return type. Hadoop's `getFactoryMethod` also requires
`returnType.isAssignableFrom(m.getReturnType())`, which is what stops an
unrelated static `create()` or `getInstance()` from being invoked and only
failing at the cast. Worth matching?
##########
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:
No objection to the mechanism, this matches every other MinIO suite we have.
It does mean the one test that reproduces #6022 end to end does not run
anywhere. Did you get a chance to run it locally, and on which profile? Worth
saying so in the description, since the existing review notes MinIO was not
exercised and the description leads with that test as the evidence.
##########
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:
Given the chain fall-through I described on the Rust side, I do not think
assumed-role belongs in this list, and "accepts everything the chain accepts"
is stronger than what the adapter actually does.
The sentence further down frames the static-key gap as narrow, but because
`AWSCredentialProviderList` skips a provider that throws rather than failing,
it is both broader and silent. Could this say plainly that any provider in the
chain which reads the static keys or the session token will resolve nothing and
the chain will move on to the next one? That is the part someone is going to
get bitten by.
--
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]
