comphead commented on code in PR #7026:
URL: https://github.com/apache/datafusion-comet/pull/7026#discussion_r4332013918
##########
spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala:
##########
@@ -406,16 +406,32 @@ object NativeConfig {
* container and account of one URI and forwarded under global key names, so
a scan whose ABFS
* files span more than one container or account, or that mixes ABFS and
other schemes, is
* declined, and one that stays in a single container forwards the
resolution of its first file.
+ *
+ * An S3 scan is also declined when the native S3 client would resolve a
different configuration
+ * than Hadoop's `S3AFileSystem` for any of its buckets (see
[[S3ConfigGate]]), since the
+ * forwarded options would then make native read under a different effective
identity or
+ * endpoint than Spark would, silently. A caller whose own claim gate
already ran
+ * [[S3ConfigGate.declineReason]] over a superset of `uris` passes
`verifyS3Config = false`
+ * rather than pay for the check twice.
*/
def extractScanObjectStoreOptions(
hadoopConf: Configuration,
- uris: Iterable[URI]): Either[String, Map[String, String]] = {
+ uris: Iterable[URI],
+ verifyS3Config: Boolean = true): Either[String, Map[String, String]] = {
val iter = uris.iterator
if (!iter.hasNext) {
return Right(Map.empty)
}
val first = iter.next()
val firstIsAbfs = isAbfs(first)
+ // One representative URI per S3 bucket: the gate decides per bucket, so
the files of a large
+ // scan need not each be re-examined by every gate.
+ val s3Uris = scala.collection.mutable.LinkedHashMap.empty[String, URI]
+ def collectS3(uri: URI): Unit =
+ if (verifyS3Config) {
+ S3ConfigGate.s3Bucket(uri).foreach(bucket =>
s3Uris.getOrElseUpdate(bucket, uri))
Review Comment:
This collects `s3` and `s3a` URIs even when native reads them through
libhdfs. For a scheme listed in `fs.comet.libhdfs.schemes`, the store is built
with `create_hdfs_object_store` and only `fs.comet.*` options are kept, so none
of the `fs.s3a.*` settings this gate compares are read. Routing `s3a` through
libhdfs is how a deployment gets Hadoop's own `S3AFileSystem` for the settings
this gate declines (a proxy, SSE-C, an unrecognized provider class), and such
scans would now fall back to Spark. `extractObjectStoreOptions` further down
already treats libhdfs URIs separately through `objectStoreKey(...).isLibhdfs`.
Could this skip URIs whose scheme is in `resolveLibhdfsSchemes(hadoopConf)`,
with a `NativeConfigSuite` case using `fs.comet.libhdfs.schemes=s3a` and
`fs.s3a.proxy.host`? Not run.
##########
docs/source/user-guide/latest/datasources.md:
##########
@@ -424,7 +424,28 @@ Comet's S3 support has the following limitations:
2. **Custom credential providers**: Custom credential provider classes named
in `fs.s3a.aws.credentials.provider` are not supported; only the standard
providers listed in the table above are. To route credential requests through
your own Java code, implement Comet's `CometS3CredentialProvider` SPI; see [S3
Credential Providers](s3-credential-providers.md). Broader Hadoop S3A
integration is tracked in
[#1829](https://github.com/apache/datafusion-comet/issues/1829).
-3. **Settings that Hadoop's S3A client resolves differently**: The native
scans resolve storage settings with their own client rather than with Hadoop's.
Where the two disagree, a native read can use a different identity or route
than Spark and fail where Spark succeeds. Examples are JCEKS-backed secrets and
the precedence of per-bucket settings, which the native Parquet scan forwards
without checking for a difference
([#5662](https://github.com/apache/datafusion-comet/issues/5662)).
+3. **S3 settings that Hadoop and Comet resolve differently**: Comet does not
run Hadoop's `S3AFileSystem`, so some
+ settings take effect differently in the native client than they do for
Spark's reader. To keep a scan from
+ silently reading under a different identity or endpoint than Spark would,
Comet declines the scan, and Spark
Review Comment:
This reads as covering every native S3 read, but the gate only runs through
`extractScanObjectStoreOptions`, which covers the native Parquet, CSV and Delta
scans. The native Iceberg scan, native writes and opted-in
`fs.comet.s3Compliant.schemes` aliases still forward `fs.s3a.*` through
`extractObjectStoreOptions` without it. The old text kept that visible through
the #5662 link this edit removes, and #5662 asks for the check on every native
read. Could this item name the scans it covers and keep a link or a follow-up
issue for the rest?
##########
spark/src/main/scala/org/apache/comet/objectstore/S3ConfigGate.scala:
##########
@@ -0,0 +1,1250 @@
+/*
+ * 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.objectstore
+
+import java.io.IOException
+import java.net.{URI, URISyntaxException}
+import java.util.Locale
+
+import scala.collection.mutable.{Map => MutableMap}
+import scala.jdk.CollectionConverters._
+
+import org.apache.hadoop.conf.Configuration
+
+/**
+ * Decides whether Comet's native S3 client would resolve the same S3
configuration as Hadoop's
+ * `S3AFileSystem` for the buckets a scan touches. The native client builds
its configuration from
+ * the options [[NativeConfig.extractObjectStoreOptions]] forwards, and
resolves them differently
+ * from Hadoop in several ways (`lookupPassword` bucket precedence, JCEKS
credential provider
+ * aliases, the clear text fallback, assumed role session policies, provider
class semantics,
+ * scheme-less endpoints). A scan that went native over such a table would
read under a different
+ * effective identity or endpoint than Spark would, silently. Each gate here
returns a decline
+ * reason for the scan to fall back to Spark, or `None` when native and Hadoop
agree.
+ *
+ * Shared by every native scan that forwards `fs.s3a.*` options (see
+ * [[NativeConfig.extractScanObjectStoreOptions]]) and by the Delta contrib's
own claim gates.
+ * Correctness rule: when in doubt, decline.
+ */
+object S3ConfigGate {
+
+ /**
+ * First reason the native S3 client would not resolve the same
configuration as Hadoop for any
+ * bucket among `uris`, or `None` when it would, or when no URI is an
`s3`/`s3a` one. The gates
+ * are ordered cheapest and most specific first. They share one
propagated-conf cache so each
+ * bucket's full `Configuration` deep copy is made at most once.
+ */
+ def declineReason(hadoopConf: Configuration, uris: Seq[URI]): Option[String]
= {
+ if (!uris.exists(uri => s3Bucket(uri).isDefined)) {
+ return None
+ }
+ val propagatedConfCache = MutableMap.empty[String, Configuration]
+ unsupportedEncryptionAlgorithmReason(hadoopConf, uris, propagatedConfCache)
+ .orElse(proxyGateReason(hadoopConf, uris, propagatedConfCache))
+ .orElse(assumedRolePolicyGateReason(hadoopConf, uris,
propagatedConfCache))
+ .orElse(hadoopOnlyEndpointGateReason(hadoopConf, uris,
propagatedConfCache))
+ .orElse(s3ConfigDivergenceReason(hadoopConf, uris, propagatedConfCache))
+ .orElse(providerClassGateReason(hadoopConf, uris))
+ }
+
+ /**
+ * String-literal Hadoop conf keys consulted below. `hadoop-aws` is NOT on
this module's runtime
+ * classpath, so `org.apache.hadoop.fs.s3a.Constants` must never be
referenced here (would raise
+ * `NoClassDefFoundError` for sessions with no S3 dependency).
+ */
+ private val HadoopCredentialProviderPathKey =
"hadoop.security.credential.provider.path"
+ private val S3aCredentialProviderPathKey =
"fs.s3a.security.credential.provider.path"
+
+ /**
+ *
`CommonConfigurationKeysPublic.HADOOP_SECURITY_CREDENTIAL_CLEAR_TEXT_FALLBACK`,
default
+ * `true`, verified via `javap` against `hadoop-common` 3.3.4's
+ * `Configuration#getPasswordFromConfig`: `getPassword` only falls back to
reading a plaintext
+ * conf value once `getBoolean(<this key>, true)` holds -- with the flag
off, a plaintext value
+ * is invisible to every `getPassword`-based resolver, even when no
credential provider is
+ * configured at all.
+ */
+ private val ClearTextFallbackKey =
"hadoop.security.credential.clear-text-fallback"
+
+ private def s3aBucketProviderPathKey(bucket: String): String =
+ s"fs.s3a.bucket.$bucket.security.credential.provider.path"
+
+ /**
+ * The LONG form of [[s3aBucketProviderPathKey]]: `S3AUtils#lookupPassword`
resolves per-bucket
+ * overrides through both a long key (`fs.s3a.bucket.B.<full base key>`) and
a short key; both
+ * must be covered here too.
+ */
+ private def s3aBucketLongProviderPathKey(bucket: String): String =
+ s"fs.s3a.bucket.$bucket.fs.s3a.security.credential.provider.path"
+
+ private def nonEmptyConf(hadoopConf: Configuration, key: String): Boolean =
+ Option(hadoopConf.get(key)).exists(_.nonEmpty)
+
+ /**
+ * The lowercase-scheme-checked S3/S3A bucket name from `uri`'s authority,
or `None` when
+ * `uri`'s scheme is not `s3`/`s3a`. Parses the raw authority manually
rather than
+ * `URI#getHost`, avoiding the RFC 3986 `reg-name` pitfall (`getHost`
returns `null` for the
+ * whole authority).
+ */
+ private[comet] def s3Bucket(uri: URI): Option[String] = {
+ val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT))
+ if (scheme.contains("s3") || scheme.contains("s3a")) {
+ val authority = Option(uri.getAuthority).getOrElse("")
+ val at = authority.lastIndexOf('@')
+ val hostAndPort = if (at >= 0) authority.substring(at + 1) else authority
+ val colon = hostAndPort.lastIndexOf(':')
+ val host = if (colon >= 0) hostAndPort.substring(0, colon) else
hostAndPort
+ if (host.isEmpty) None else Some(host)
+ } else {
+ None
+ }
+ }
+
+ private def plainValue(hadoopConf: Configuration, key: String):
Option[String] =
+ Option(hadoopConf.get(key)).filter(_.nonEmpty)
+
+ /**
+ * How Hadoop's OWN consumer reads one of the keys compared by
[[s3ConfigDivergenceReason]],
+ * which decides how [[s3KeyDivergenceReason]] computes the Hadoop-effective
side of its
+ * equality check. Exactly two consumer families exist among
[[AllS3ConfigKeys]] in `hadoop-aws`
+ * 3.3.4, each verified via `javap`/CFR against the real call sites (cited
per key on
+ * [[S3ConfigKeyConsumers]]). The tier must mirror the key's ACTUAL
consumer: resolving a
+ * [[PropagatedOptionConsumer]] key through the wider `lookupPassword`
cascade is NOT fail-safe
+ * for a value-EQUALITY comparator -- a long-form alias value Hadoop itself
never reads can
+ * EQUAL native's resolution while Hadoop's true propagate-then-plain-get
value differs, turning
+ * a real divergence into a wrongly-claimed scan (the endpoint
`${...}`-redirect shape pinned in
+ * `S3ConfigGateSuite`).
+ */
+ private[comet] sealed trait S3ConfigConsumer
+
+ /**
+ * Read via `S3AUtils#lookupPassword(bucket, conf, baseKey)`, verified via
`javap` against
+ * `hadoop-aws` 3.3.4: builds `longBucketKey = "fs.s3a.bucket." + bucket +
"." + baseKey` (the
+ * FULL, already-`fs.s3a`-prefixed base key appended after the bucket
segment) and reads it via
+ * `Configuration#getPassword` BEFORE the short-bucket key, keeping the long
value whenever
+ * `getPassword` returns non-empty and only falling through to
short-then-global otherwise.
+ * `getPassword` is Hadoop-credential-provider-aware and skips plaintext
conf entirely when
+ * [[ClearTextFallbackKey]] is false. Modeled by
[[hadoopLookupPasswordEffective]].
+ */
+ private[comet] case object LookupPasswordConsumer extends S3ConfigConsumer
+
+ /**
+ * Read via `S3AUtils#propagateBucketOptions` followed by a plain
`Configuration#get`-family
+ * call (`getTrimmed`/`getBoolean`/`getClasses`) against the propagated
view: the short bucket
+ * form wins only by having overwritten the global key during propagation,
the long bucket form
+ * folds into an unread `fs.s3a.fs.s3a.*` key, and neither a credential
provider nor
+ * [[ClearTextFallbackKey]] is ever consulted. Modeled as a plain
`Configuration#get` on the
+ * [[propagateBucketOptions]] result, which also expands `${...}` references
under that
+ * propagated view exactly like the real consumer.
+ */
+ private[comet] case object PropagatedOptionConsumer extends S3ConfigConsumer
+
+ /**
+ * Every `fs.s3a.*` base key that governs whether a claimed native scan
actually behaves like
+ * Hadoop's own reader would, paired with the consumer family Hadoop
resolves it through -- ONE
+ * list, with each key's resolution tier declared beside it, so a key can
never sit in the
+ * comparator without a deliberate classification (adding one without
picking a tier does not
+ * compile). The entries are every per-bucket `fs.s3a.*` base key native's
S3 client's
+ * `get_config` (s3.rs) resolves, verified directly against its call sites:
+ * `extract_s3_config_options` (endpoint.region, path.style.access, endpoint,
+ * requester.pays.enabled), `lookup_provider_class` (the Comet-specific
+ * credential-provider-class activation key), and
+ * `build_credential_provider`/`build_aws_credential_provider_metadata`/
+ * `build_assume_role_credential_provider_metadata`
(aws.credentials.provider,
+ * assumed.role.credentials.provider, assumed.role.arn,
assumed.role.session.name).
+ *
+ * Tier assignments, each verified via `javap`/CFR against `hadoop-aws`
3.3.4:
+ * - access.key/secret.key/session.token: `S3AUtils#getAWSAccessKeys` and
+ * `MarshalledCredentialBinding#fromFileSystem` (reached from
+ * `TemporaryAWSCredentialsProvider`) resolve all three via
`S3AUtils#lookupPassword` --
+ * [[LookupPasswordConsumer]].
+ * - aws.credentials.provider and assumed.role.credentials.provider:
+ * `S3AUtils#buildAWSProviderList` -> `loadAWSProviderClasses` -> plain
+ * `Configuration#getClasses` -- [[PropagatedOptionConsumer]].
+ * - assumed.role.arn/session.name: `AssumedRoleCredentialProvider`'s
constructor reads both
+ * via plain `Configuration#getTrimmed` -- [[PropagatedOptionConsumer]].
+ * - endpoint (`S3AFileSystem`: `getTrimmed`), endpoint.region
(`DefaultS3ClientFactory`:
+ * `getTrimmed`), path.style.access (`S3AFileSystem`: `getBoolean`) --
+ * [[PropagatedOptionConsumer]].
+ * - requester.pays.enabled: not read anywhere in `hadoop-aws` 3.3.4 (the
constant does not
+ * even exist in its `Constants` class); later releases read it via
plain `getBoolean`
+ * against the propagated conf, so the plain tier is both the faithful
forward model and
+ * inert on 3.3.4 -- [[PropagatedOptionConsumer]].
+ * - comet.credential.provider.class: Comet's own activation key, plain
conf read on both
+ * sides, never a Hadoop key at all -- [[PropagatedOptionConsumer]].
+ *
+ * SYNC NOTE: the key list must stay a superset of native's
`NATIVE_S3A_CONFIG_PROPERTIES`
+ * constant (`native/core/src/parquet/objectstore/s3.rs`, property suffixes
without the
+ * `fs.s3a.` prefix) -- `S3ConfigGateSuite`'s discovery-harness test asserts
this mechanically
+ * against [[AllS3ConfigKeys]]. A Rust test in `s3.rs` also parses this
`Seq(` block as text, so
+ * each entry must stay a single `"fs.s3a.<key>" -> <Tier>` literal pair.
Literal strings, not
+ * the [[AwsCredentialsProviderKey]] / [[AssumedRoleCredentialsProviderKey]]
vals declared
+ * below, purely to avoid a forward reference inside this `object` body;
kept textually
+ * identical to those two constants.
+ */
+ private[comet] val S3ConfigKeyConsumers: Seq[(String, S3ConfigConsumer)] =
Seq(
+ "fs.s3a.access.key" -> LookupPasswordConsumer,
+ "fs.s3a.secret.key" -> LookupPasswordConsumer,
+ "fs.s3a.session.token" -> LookupPasswordConsumer,
+ "fs.s3a.aws.credentials.provider" -> PropagatedOptionConsumer,
+ "fs.s3a.assumed.role.arn" -> PropagatedOptionConsumer,
+ "fs.s3a.assumed.role.session.name" -> PropagatedOptionConsumer,
+ "fs.s3a.assumed.role.credentials.provider" -> PropagatedOptionConsumer,
+ "fs.s3a.endpoint" -> PropagatedOptionConsumer,
+ "fs.s3a.endpoint.region" -> PropagatedOptionConsumer,
+ "fs.s3a.path.style.access" -> PropagatedOptionConsumer,
+ "fs.s3a.requester.pays.enabled" -> PropagatedOptionConsumer,
+ "fs.s3a.comet.credential.provider.class" -> PropagatedOptionConsumer)
+
+ /** The compared keys alone, in [[S3ConfigKeyConsumers]] order
(discovery-harness surface). */
+ private[comet] val AllS3ConfigKeys: Seq[String] =
S3ConfigKeyConsumers.map(_._1)
+
+ /**
+ * The short-bucket-then-global value resolved for `baseKey` under `bucket`
from `hadoopConf`,
+ * skipping an empty value at either alias exactly like [[plainValue]]. NOT
used by
+ * [[s3ConfigDivergenceReason]]/[[s3KeyDivergenceReason]] any more -- every
key checked there
+ * resolves per its declared [[S3ConfigKeyConsumers]] tier (see
[[s3KeyDivergenceReason]]),
+ * neither of which matches this function's read. This function's one
remaining caller is
+ * [[shortThenGlobalOrReason]], which reads provider-CLASS strings (from the
ORIGINAL,
+ * unpropagated conf) for name-support validation in [[providerClassReason]]/
+ * [[assumedRoleProviderClassReason]] -- by the time those run,
[[s3ConfigDivergenceReason]] has
+ * already proven Hadoop's and native's effective values agree for the same
key, so whichever of
+ * the two (equal) values this narrower read returns does not affect
correctness there. NEVER
+ * used to compute native's own effective value -- see
[[nativeShortThenGlobal]] for that.
+ */
+ private def shortThenGlobal(
+ hadoopConf: Configuration,
+ bucket: String,
+ baseKey: String): Option[String] = {
+ val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.")
+ plainValue(hadoopConf, shortKey).orElse(plainValue(hadoopConf, baseKey))
+ }
+
+ /**
+ * The short-bucket-then-global value native's `get_config` (s3.rs) resolves
for `baseKey` under
+ * `bucket` from the ORIGINAL, unpropagated `hadoopConf` -- `NativeConfig
+ * .extractObjectStoreOptions` forwards `Configuration#get`'s substituted
value for every
+ * `fs.s3a.*` entry with no bucket-option propagation step of its own, so
the original conf is
+ * the right input here. Unlike [[shortThenGlobal]]/[[plainValue]], this
mirrors `get_config`
+ * faithfully: PRESENCE of the short-bucket key alone -- never its emptiness
-- decides whether
+ * native falls back to the global key (`get_config` is a plain
`HashMap::get`, which returns
+ * `Some` for a key explicitly set to `""`), so an explicitly empty or
whitespace-only
+ * short-bucket value resolves to `Some("")` here and never falls through to
global -- the
+ * OPPOSITE of Hadoop's own `getPassword`/`lookupPassword` semantics (see
+ * [[hadoopLookupPasswordEffective]]), which treat empty as absent and keep
trying the next
+ * alias. The ONLY function used to compute native's effective value in
+ * [[s3KeyDivergenceReason]].
+ *
+ * Deliberately does NOT apply `get_config_trimmed`'s `.trim()` here:
[[s3KeyDivergenceReason]]
+ * trims both this value and Hadoop's effective value together,
symmetrically, at the point they
+ * are compared, rather than one-sidedly here -- trimming only the native
side would flag a
+ * spurious divergence for a value neither side's whitespace actually
changes the behavior of
+ * once each side's own downstream parsing normalizes it (e.g. Hadoop's own
multi-line
+ * `fs.s3a.aws.credentials.provider` default, which both Hadoop and native
additionally trim per
+ * comma-separated entry after splitting), while a one-sided trim would make
an
+ * otherwise-identical default value look diverged for every bucket, never
claiming natively at
+ * all.
+ */
+ private def nativeShortThenGlobal(
+ hadoopConf: Configuration,
+ bucket: String,
+ baseKey: String): Option[String] = {
+ val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.")
+ Option(hadoopConf.get(shortKey)).orElse(Option(hadoopConf.get(baseKey)))
+ }
+
+ /**
+ * Faithful in-memory replica of `S3AUtils#propagateBucketOptions`
(`hadoop-aws`), which
+ * `S3AFileSystem#initialize` calls FIRST, before any option or credential
is read:
+ * `Configuration conf = propagateBucketOptions(originalConf, bucket); ...;
setConf(conf);` --
+ * every subsequent `conf.get`/`getPassword` call in that filesystem
instance, including
+ * `${...}` variable substitution, resolves against this propagated view,
not the original conf.
+ * `hadoop-aws` is not on this module's runtime classpath (see the
string-literal-keys note
+ * above), so `S3AUtils#propagateBucketOptions` cannot be called directly;
this reproduces its
+ * logic verbatim using only `hadoop-common`'s `Configuration`:
+ * {{{
+ * public static Configuration propagateBucketOptions(Configuration source,
String bucket) {
+ * final String bucketPrefix = FS_S3A_BUCKET_PREFIX + bucket + '.';
+ * final Configuration dest = new Configuration(source);
+ * for (Map.Entry<String, String> entry : source) {
+ * final String key = entry.getKey();
+ * final String value = entry.getValue(); // the (unexpanded) value
+ * if (!key.startsWith(bucketPrefix) || bucketPrefix.equals(key))
continue;
+ * final String stripped = key.substring(bucketPrefix.length());
+ * if (stripped.startsWith("bucket.") || "impl".equals(stripped)) {
+ * // ignored
+ * } else {
+ * final String generic = FS_S3A_PREFIX + stripped;
+ * dest.set(generic, value, ...); // overwrites any existing global
value
+ * }
+ * }
+ * return dest;
+ * }
+ * }}}
+ * Note the LONG bucket form (`fs.s3a.bucket.B.fs.s3a.<key>`) folds to an
unread
+ * `fs.s3a.fs.s3a.<key>` key here too, exactly like the real method --
`stripped` already starts
+ * with `fs.s3a.` in that case, so prepending `fs.s3a.` again produces a key
nothing ever reads.
+ */
+ private def propagateBucketOptions(hadoopConf: Configuration, bucket:
String): Configuration = {
+ val bucketPrefix = s"fs.s3a.bucket.$bucket."
+ val dest = new Configuration(hadoopConf)
+ hadoopConf.iterator().asScala.foreach { entry =>
+ val key = entry.getKey
+ if (key.startsWith(bucketPrefix) && key != bucketPrefix) {
+ val stripped = key.substring(bucketPrefix.length)
+ if (!stripped.startsWith("bucket.") && stripped != "impl") {
+ dest.set(s"fs.s3a.$stripped", entry.getValue)
+ }
+ }
+ }
+ dest
+ }
+
+ /**
+ * Canonical and deprecated Hadoop S3A encryption-algorithm config keys,
verified via `javap`
+ * against `hadoop-aws` 3.3.4's `org.apache.hadoop.fs.s3a.Constants`:
`S3_ENCRYPTION_ALGORITHM =
+ * "fs.s3a.encryption.algorithm"` (canonical) and
`SERVER_SIDE_ENCRYPTION_ALGORITHM =
+ * "fs.s3a.server-side-encryption-algorithm"` (DEPRECATED -- note the hyphen
before "algorithm",
+ * unlike the corresponding `*.key` constants below, which both use a `.key`
suffix).
+ * `hadoop-aws` is NOT on this module's runtime classpath, so these stay
string literals, same
+ * rationale as [[HadoopCredentialProviderPathKey]].
+ */
+ private val S3EncryptionAlgorithmKey = "fs.s3a.encryption.algorithm"
+ private val DeprecatedS3EncryptionAlgorithmKey =
"fs.s3a.server-side-encryption-algorithm"
+
+ /**
+ * The exact strings `S3AEncryptionMethods#getMethod` accepts, verified via
`javap`/CFR against
+ * `hadoop-aws` 3.3.4's `S3AEncryptionMethods` enum: `NONE("")`,
`SSE_S3("AES256", serverSide =
+ * true, requiresSecret = false)`, `SSE_KMS("SSE-KMS", serverSide = true,
requiresSecret =
+ * false)`, `SSE_C("SSE-C", serverSide = true, requiresSecret = true)`,
`CSE_KMS("CSE-KMS",
+ * serverSide = false, requiresSecret = true)`, `CSE_CUSTOM("CSE-CUSTOM",
serverSide = false,
+ * requiresSecret = true)`. `getMethod` parses case-insensitively
+ * (`values().find(_.getMethod.equalsIgnoreCase(algorithm))`), matched below
the same way.
+ *
+ * ALLOWLIST, not a blocklist (replaces the former SSE-C-only blocklist):
only the algorithms S3
+ * decrypts transparently on GET/HEAD given read permission alone, with NO
extra request header
+ * and NO client-side step, are safe for a native scan that forwards none of
Hadoop's
+ * `fs.s3a.encryption.*`/`fs.s3a.server-side-encryption*` options --
+ * - `AES256` (SSE_S3, `serverSide = true`): plain server-side encryption,
transparent on GET.
+ * - `SSE-KMS` (SSE_KMS, `serverSide = true`): server-side, KMS-managed
key, transparent on
+ * GET given KMS decrypt permission (no header).
+ * - `DSSE-KMS`: NOT present in this enum on `hadoop-aws` 3.3.4 (confirmed
by the six values
+ * listed above) -- `S3AEncryptionMethods.getMethod("DSSE-KMS")` throws
+ * `IOException("Unknown encryption algorithm DSSE-KMS")` on this
version, so
+ * `S3AUtils#buildEncryptionSecrets` (and therefore Hadoop's own reader)
already fails
+ * before ever reading such a table under 3.3.4, meaning this string can
never actually be
+ * the resolved value on the declared target version -- admitting it
here is inert there.
+ * Included anyway, forward-compatible, for a newer `hadoop-aws` on the
runtime classpath (a
+ * later Hadoop release; this module has no compile-time `hadoop-aws`
dependency, see the
+ * string-literal-keys note above) where DSSE-KMS is a real, dual-layer,
server-side
+ * algorithm decrypted transparently on GET the same way SSE-KMS is.
Every other value
+ * declines: `SSE-C` (SSE_C is `serverSide = true` in Hadoop's own enum,
but `requiresSecret
+ * \= true` -- S3 rejects a GET/HEAD for an SSE-C object outright (400
Bad Request) unless
+ * the customer key is resent as a request header on every call, so a
native scan that never
+ * learns the key cannot succeed at all, where Hadoop's own reader --
whose request factory
+ * attaches the key -- would), `CSE-KMS`/`CSE-CUSTOM` (`serverSide =
false`: client-side
+ * encryption decrypts object bytes locally in the SDK layer, which the
native Parquet
+ * reader has no equivalent of -- it would read raw ciphertext), and any
future/unknown
+ * value (a value `S3AEncryptionMethods.getMethod` itself would reject
is certainly not one
+ * of the three confirmed-transparent algorithms above; declining is the
only safe default
+ * for anything this gate cannot positively confirm).
+ */
+ private val AllowedEncryptionAlgorithms: Set[String] = Set("AES256",
"SSE-KMS", "DSSE-KMS")
+
+ /**
+ * `bucket`'s effective encryption-algorithm key and value under
`hadoopConf`, or `None` when
+ * neither the canonical nor deprecated key is set anywhere consulted.
Mirrors
+ * `S3AUtils#buildEncryptionSecrets`'s real resolution order, verified via
`javap`/CFR
+ * decompilation of `hadoop-aws` 3.3.4's `S3AUtils.class`:
+ * {{{
+ * String algorithm = lookupBucketSecret(bucket, conf,
"fs.s3a.encryption.algorithm");
+ * if (algorithm == null)
+ * algorithm = lookupBucketSecret(bucket, conf,
"fs.s3a.server-side-encryption-algorithm");
+ * if (algorithm == null)
+ * algorithm = lookupPassword(null, conf, "fs.s3a.encryption.algorithm");
+ * if (algorithm == null)
+ * algorithm = lookupPassword(null, conf,
"fs.s3a.server-side-encryption-algorithm");
+ * }}}
+ * i.e. bucket-tier (canonical, then deprecated), THEN global-tier
(canonical, then deprecated)
+ * -- the two tiers are never interleaved key-by-key, so this must stay two
explicit bucket-tier
+ * lookups followed by two explicit global-tier lookups, not a single
+ * [[hadoopLookupPasswordEffective]] call per key (which would let an unset
canonical bucket key
+ * fall through straight to the canonical GLOBAL value ahead of a SET
deprecated bucket key, the
+ * wrong answer).
+ *
+ * The three-argument `lookupPassword` defaults to `""`, not `null`, so the
last call in the
+ * block above never runs: Hadoop reads the deprecated global key only
through `Configuration`'s
+ * deprecation alias of the canonical one. This function still reads it as a
fourth lookup,
+ * which can only add a decline.
+ *
+ * THE FIX for the SSE-C long-bucket-alias gap is entirely inside the bucket
tier:
+ * `lookupBucketSecret` itself is long-then-short, decompiled from
`hadoop-aws` 3.3.4's
+ * `S3AUtils.class`:
+ * {{{
+ * // longBucketKey = fs.s3a.bucket.B.fs.s3a.<key>
+ * String longBucketKey = String.format(BUCKET_PATTERN, bucket, baseKey);
+ * String initialVal = getPassword(conf, longBucketKey, null, null);
+ * // shortBucketKey = fs.s3a.bucket.B.<key>
+ * String shortBucketKey = String.format(BUCKET_PATTERN, bucket, subkey);
+ * // keeps initialVal (the LONG value) if non-empty
+ * return getPassword(conf, shortBucketKey, initialVal, null);
+ * }}}
+ * i.e. the SAME long-bucket-key construction and long-wins-if-nonempty
semantics as
+ * `S3AUtils#lookupPassword` (see
[[LookupPasswordConsumer]]/[[hadoopLookupPasswordEffective]]).
+ * `buildEncryptionSecrets` runs on the [[propagateBucketOptions]] view, so
+ * [[unsupportedEncryptionAlgorithmReason]] calls this on that view (and on
the raw conf, which
+ * can only add a decline). An earlier version of this function modeled the
bucket tier as
+ * SHORT-only, documented as "the LONG bucket form is genuinely never
consulted for this key" --
+ * that documentation was wrong (this decompilation supersedes it): a bucket
configured only via
+ * `fs.s3a.bucket.B.fs.s3a.encryption.algorithm=SSE-C` bypassed the SSE-C
gate entirely, because
+ * Hadoop's own reader DOES read that long form (and picks SSE-C), while
this function reported
+ * `None` (nothing set) and the allowlist check below never even ran.
+ *
+ * The canonical-vs-deprecated distinction below is frequently moot in
practice: `hadoop-aws`'s
+ * `S3AFileSystem.addDeprecatedKeys()` statically registers
`fs.s3a.server-side-encryption-*` as
+ * `Configuration`-level deprecated aliases of `fs.s3a.encryption.*`
(verified via `javap`), a
+ * registration that lives in a static field on Hadoop's `Configuration`
class -- process-wide
+ * once `S3AFileSystem`'s class has loaded anywhere in the JVM, which a real
scan has always
+ * already done by the time this gate runs, since reading the S3 table at
all requires loading
+ * that class. Once active, `Configuration#get` resolves either literal key
to the identical
+ * value transparently, making the two-key cascade below redundant (but
harmless) for that case;
+ * it remains the operative path only when nothing else in the process has
loaded
+ * `S3AFileSystem` yet.
+ *
+ * ALSO walks the Hadoop-credential-provider (JCEKS) path via
[[resolveViaCredentialAliases]]
+ * for each of the four lookups below, matching
`lookupBucketSecret`/`lookupPassword`'s real
+ * per-alias `getPassword` calls (quoted above) exactly: both are
`getPassword`, not plain
+ * `Configuration#get`, so a bucket storing the algorithm name ONLY in a
JCEKS keystore is
+ * exactly as real a Hadoop deployment shape for this key as it is for the
credential keys
+ * [[hadoopLookupPasswordEffective]] already covers -- there is nothing
algorithm-specific that
+ * makes JCEKS storage implausible here, so an earlier version of this
function skipping it
+ * (documented at the time as "the algorithm NAME is not
credential-sensitive data, so storing
+ * it in a keystore is not a realistic Hadoop deployment pattern") was an
unjustified, narrower
+ * read than Hadoop's own resolver actually performs, under-declining a
bucket whose algorithm
+ * is keystore-only. [[resolveViaCredentialAliases]]'s Arm B/C split still
means this is zero
+ * extra I/O for the common case: keystore I/O only happens when a Hadoop
credential-provider
+ * path is actually configured for the bucket, contained in that function's
own try/catch.
+ * `bucketTier`/`globalTier` return `Left` (propagated straight through by
[[orElseTier]]) when
+ * [[resolveViaCredentialAliases]] cannot safely verify a tier at all (an
S3A-scoped provider
+ * path, or a corrupt/unreadable global keystore) -- correctly
short-circuiting the whole
+ * cascade with a decline rather than silently falling through to a later
tier that might look
+ * unset only because the true value was unverifiable.
+ */
+ private def effectiveEncryptionAlgorithm(
+ hadoopConf: Configuration,
+ bucket: String): Either[String, Option[(String, String)]] = {
+ def bucketTier(baseKey: String): Either[String, Option[(String, String)]]
= {
+ val longKey = s"fs.s3a.bucket.$bucket.$baseKey"
+ val shortKey = s"fs.s3a.bucket.$bucket." + baseKey.stripPrefix("fs.s3a.")
+ resolveViaCredentialAliases(hadoopConf, bucket, Seq(longKey, shortKey))
+ .map(_.map(baseKey -> _))
+ }
+ def globalTier(baseKey: String): Either[String, Option[(String, String)]] =
+ resolveViaCredentialAliases(hadoopConf, bucket, Seq(baseKey))
+ .map(_.map(baseKey -> _))
+
+ // Short-circuits on Left (unverifiable tier) or Right(Some(_))
(resolved); only Right(None)
+ // (tier definitively unset) falls through to `next`, mirroring
buildEncryptionSecrets's
+ // sequential `if (algorithm == null) algorithm = ...` cascade exactly.
+ def orElseTier(
+ current: Either[String, Option[(String, String)]],
+ next: => Either[String, Option[(String, String)]])
+ : Either[String, Option[(String, String)]] =
+ current match {
+ case Left(reason) => Left(reason)
+ case Right(Some(value)) => Right(Some(value))
+ case Right(None) => next
+ }
+
+ orElseTier(
+ bucketTier(S3EncryptionAlgorithmKey),
+ orElseTier(
+ bucketTier(DeprecatedS3EncryptionAlgorithmKey),
+ orElseTier(
+ globalTier(S3EncryptionAlgorithmKey),
+ // Hadoop reaches this key only through the deprecation alias (see
above); reading
+ // it directly can only over-decline.
+ globalTier(DeprecatedS3EncryptionAlgorithmKey))))
+ }
+
+ /**
+ * The only algorithm values a decline reason may print. Any other resolved
value is withheld,
+ * since a `${...}` reference can expand the algorithm key to a customer key
or token.
+ */
+ private val PrintableEncryptionAlgorithms: Seq[String] =
+ Seq("AES256", "SSE-S3", "SSE-KMS", "DSSE-KMS", "SSE-C", "CSE-KMS",
"CSE-CUSTOM")
+
+ private def unsupportedEncryptionAlgorithmDeclineReason(
+ bucket: String,
+ algorithmKey: String,
+ algorithm: String): String = {
+ val setting = PrintableEncryptionAlgorithms
+ .find(_.equalsIgnoreCase(algorithm))
+ .map(name => s"$algorithmKey=$name")
+ .getOrElse(s"$algorithmKey set to a value that is not a recognised S3
encryption algorithm")
+ s"Native scan does not support $setting for $bucket " +
+ "(the native S3 client only supports unencrypted objects and S3's
transparent " +
+ "server-side algorithms -- AES256/SSE-S3, SSE-KMS, and DSSE-KMS decrypt
on GET/HEAD given " +
+ "read permission alone, with no extra request header; SSE-C additionally
requires the " +
+ "customer-provided key resent as a header on every GET/HEAD request,
which the native S3 " +
+ "client's extract_s3_config_options never forwards, and
CSE-KMS/CSE-CUSTOM decrypt object " +
+ "bytes client-side, a layer the native Parquet reader does not have --
any of these would " +
+ "fail outright or silently read ciphertext where Hadoop's own reader
succeeds)"
+ }
+
+ /**
+ * First reason any bucket among `uris` is configured for an encryption
algorithm the native S3
+ * client cannot safely read, or `None` when claimable. Allowlist-based (see
+ * [[AllowedEncryptionAlgorithms]]): only `AES256`/`SSE-KMS`/`DSSE-KMS` (and
unset/empty) pass;
+ * every other resolved value -- `SSE-C`, `CSE-KMS`, `CSE-CUSTOM`, or any
unrecognized future
+ * algorithm string -- declines. Deliberately NOT a blocklist keyed on
`SSE-C` alone: an
+ * allowlist is safe by construction against a Hadoop release adding a new
encryption method
+ * this gate has never heard of, where a blocklist would silently admit it.
Never interpolates a
+ * resolved key value, only key names, the bucket, and the (non-secret)
algorithm name. Declines
+ * on a `Left` from [[effectiveEncryptionAlgorithm]] too (an unverifiable
credential-provider
+ * arm, e.g. an S3A-scoped provider path or a corrupt/unreadable global
keystore) -- the
+ * algorithm cannot be ruled safe when it cannot be read at all.
+ */
+ private[comet] def unsupportedEncryptionAlgorithmReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI],
+ propagatedConfCache: MutableMap[String, Configuration] =
MutableMap.empty)
+ : Option[String] = {
+ val buckets = uris.flatMap(s3Bucket).distinct
+ buckets.foldLeft(Option.empty[String]) { (declined, bucket) =>
+ if (declined.isDefined) {
+ declined
+ } else {
+ try {
+ // Hadoop resolves the algorithm on the propagated view, where a
bucket override can
+ // change what a `${...}` reference expands to. The raw view is
checked as well, so a
+ // disagreement between the two can only decline.
+ val propagatedConf =
+ propagatedConfCache.getOrElseUpdate(
+ bucket,
+ propagateBucketOptions(hadoopConf, bucket))
+ encryptionAlgorithmReason(propagatedConf, bucket)
+ .orElse(encryptionAlgorithmReason(hadoopConf, bucket))
+ } catch {
+ case e @ (_: IOException | _: RuntimeException) =>
+ Some(unverifiableValueReason(bucket, S3EncryptionAlgorithmKey, e))
+ }
+ }
+ }
+ }
+
+ private def encryptionAlgorithmReason(conf: Configuration, bucket: String):
Option[String] =
+ effectiveEncryptionAlgorithm(conf, bucket) match {
+ case Left(reason) => Some(reason)
+ case Right(None) => None
+ case Right(Some((key, value))) =>
+ if (!AllowedEncryptionAlgorithms.exists(_.equalsIgnoreCase(value))) {
+ Some(unsupportedEncryptionAlgorithmDeclineReason(bucket, key, value))
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Canonical Hadoop S3A HTTP-proxy host config key, verified via CFR
decompilation of
+ * `hadoop-aws` 3.3.4's `S3AUtils.class` (`initProxySupport`):
+ * {{{
+ * String proxyHost = conf.getTrimmed("fs.s3a.proxy.host", "");
+ * int proxyPort = conf.getInt("fs.s3a.proxy.port", -1);
+ * if (!proxyHost.isEmpty()) {
+ * ...
+ * String proxyUsername =
+ * S3AUtils.lookupPassword(bucket, conf, "fs.s3a.proxy.username",
null, null);
+ * String proxyPassword =
+ * S3AUtils.lookupPassword(bucket, conf, "fs.s3a.proxy.password",
null, null);
+ * ...
+ * }
+ * }}}
+ * `fs.s3a.proxy.host`/`fs.s3a.proxy.port` resolve via a PLAIN,
non-bucket-scoped, non-JCEKS
+ * `Configuration#getTrimmed`/`getInt` call -- NOT `lookupPassword` --
against whatever conf
+ * `S3AFileSystem#initialize` already ran through `propagateBucketOptions`
before
+ * `createAwsConf`/`initProxySupport` ever runs; only the SIBLING
`fs.s3a.proxy.username`/
+ * `fs.s3a.proxy.password` keys go through `lookupPassword` (bucket
long/short/global,
+ * JCEKS-aware). So the host is bucket-aware only via
`propagateBucketOptions`'s short-bucket-
+ * form folding, never the long-bucket form, and never a credential-provider
read -- the SAME
+ * shape as `endpoint`/`path.style.access` ([[PropagatedOptionConsumer]], see
+ * [[S3ConfigKeyConsumers]]'s doc), not the credential family. `hadoop-aws`
3.4.x (the Spark 4.x
+ * profiles' version) moves this code to
`AWSClientConfig#createProxyConfiguration`/
+ * `#createAsyncProxyConfiguration` but keeps the exact same reads, verified
via `javap` against
+ * 3.4.2: `conf.getTrimmed("fs.s3a.proxy.host", "")` for the host,
`S3AUtils.lookupPassword` for
+ * username/password only.
+ *
+ * [[proxyGateReason]] therefore resolves the host EXACTLY like its real
consumer -- plain
+ * `Configuration#getTrimmed` on the [[propagateBucketOptions]] result, no
provider arms, no
+ * [[ClearTextFallbackKey]] handling -- rather than through the wider
`lookupPassword` cascade
+ * an earlier version reused here. The wider cascade was wrong in BOTH
directions for this key:
+ * with only the GLOBAL Hadoop provider path set and
[[ClearTextFallbackKey]] false,
+ * `getPassword` hides a plaintext host that `getTrimmed` serves to Hadoop
anyway (a missed
+ * decline, the exact bypass this gate exists to close -- native has NO
HTTP-proxy support of
+ * any kind, no `fs.s3a.proxy.*` key is read anywhere in `s3.rs`); and an
S3A-scoped provider
+ * path or a lone long-form bucket alias declined a bucket whose real
consumer can never see a
+ * host from either source (pure over-refusal -- no keystore can supply the
host to a plain
+ * `getTrimmed`, and the long form folds into the unread
`fs.s3a.fs.s3a.proxy.host`).
+ */
+ private val S3ProxyHostKey = "fs.s3a.proxy.host"
+
+ private def unsupportedProxyReason(bucket: String, key: String): String =
+ s"Native scan does not support $key configured for $bucket (the native S3
client has " +
+ "no HTTP proxy support at all -- no fs.s3a.proxy.* key is read anywhere
in its object " +
+ "store layer -- so a claimed scan would connect to S3 directly instead
of routing through " +
+ "the configured proxy, either bypassing an egress/network-segmentation
policy or simply " +
+ "failing to reach the endpoint)"
+
+ /**
+ * First reason any bucket among `uris` has an HTTP proxy configured via
[[S3ProxyHostKey]], or
+ * `None` when claimable. Reads the host EXACTLY like its real consumer (see
+ * [[S3ProxyHostKey]]'s doc): plain `Configuration#getTrimmed` on the
[[propagateBucketOptions]]
+ * result, so this gate is always zero-I/O -- no credential provider is ever
consulted for the
+ * host, because none ever supplies it to Hadoop either. Ordered alongside
+ * [[unsupportedEncryptionAlgorithmReason]] among the conf-only gates, ahead
of
+ * [[s3ConfigDivergenceReason]]. The try/catch guards `Configuration#get`'s
+ * `IllegalStateException` on a `${...}` substitution cycle, same as
[[s3KeyDivergenceReason]].
+ * Never interpolates a resolved value: the proxy HOST is not secret, but
naming it here would
+ * be a strange place to first surface it, and proxy CREDENTIALS
+ * (`fs.s3a.proxy.username`/`fs.s3a.proxy.password`, not read by this gate
at all -- the whole
+ * point of gating on the host is that a non-empty host declines before any
proxy credential
+ * would ever need to be forwarded) must never appear in a decline reason
regardless.
+ */
+ private[comet] def proxyGateReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI],
+ propagatedConfCache: MutableMap[String, Configuration] =
MutableMap.empty)
+ : Option[String] = {
+ val buckets = uris.flatMap(s3Bucket).distinct
+ buckets.foldLeft(Option.empty[String]) { (declined, bucket) =>
+ if (declined.isDefined) {
+ declined
+ } else {
+ try {
+ val propagatedConf =
+ propagatedConfCache.getOrElseUpdate(
+ bucket,
+ propagateBucketOptions(hadoopConf, bucket))
+ if (propagatedConf.getTrimmed(S3ProxyHostKey, "").nonEmpty) {
+ Some(unsupportedProxyReason(bucket, S3ProxyHostKey))
+ } else {
+ None
+ }
+ } catch {
+ case e @ (_: IOException | _: RuntimeException) =>
+ Some(unverifiableValueReason(bucket, S3ProxyHostKey, e))
+ }
+ }
+ }
+ }
+
+ /**
+ * Canonical Hadoop S3A assumed-role session-policy key.
`AssumedRoleCredentialProvider`'s
+ * constructor reads it via a plain `Configuration#getTrimmed` against the
+ * [[propagateBucketOptions]]-propagated conf (verified via `javap` against
`hadoop-aws` 3.3.4
+ * and 3.4.1: `conf.getTrimmed("fs.s3a.assumed.role.policy", "")` -- the
same consumer shape as
+ * `assumed.role.arn`/`session.name`, see [[S3ConfigKeyConsumers]]) and,
when non-empty,
+ * attaches it as the session policy of its STS AssumeRole request.
+ */
+ private val S3AssumedRolePolicyKey = "fs.s3a.assumed.role.policy"
+
+ private def assumedRolePolicyReason(bucket: String, key: String): String =
+ s"Native scan does not support $key configured for $bucket (Hadoop sends
the " +
+ "configured session policy in its STS AssumeRole request, but the native
S3 client's " +
+ "assumed-role provider never reads or forwards this key, so a claimed
scan would " +
+ "assume the role WITHOUT the configured session restriction -- silently
widening the " +
+ "effective permissions instead of failing)"
+
+ /**
+ * First reason any bucket among `uris` configures an assumed-role session
policy via
+ * [[S3AssumedRolePolicyKey]], or `None` when claimable. Hadoop's
+ * `AssumedRoleCredentialProvider` includes the policy in its AssumeRole
request; native's
+ * assumed-role provider does not, so a configured policy must decline until
native supports it.
+ * Resolved exactly like the key's real consumer (plain `getTrimmed` on the
propagated conf,
+ * mirroring [[proxyGateReason]]); declined whenever set, whether or not the
current provider
+ * chain names the assumed-role provider -- a policy that is dead config
today can become live
+ * through a provider-chain change native never re-validates. Never
interpolates the policy
+ * document itself, only the key and bucket.
+ */
+ private[comet] def assumedRolePolicyGateReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI],
+ propagatedConfCache: MutableMap[String, Configuration] =
MutableMap.empty)
+ : Option[String] = {
+ val buckets = uris.flatMap(s3Bucket).distinct
+ buckets.foldLeft(Option.empty[String]) { (declined, bucket) =>
+ if (declined.isDefined || cometProviderClassConfigured(hadoopConf,
bucket)) {
+ declined
+ } else {
+ try {
+ val propagatedConf =
+ propagatedConfCache.getOrElseUpdate(
+ bucket,
+ propagateBucketOptions(hadoopConf, bucket))
+ if (propagatedConf.getTrimmed(S3AssumedRolePolicyKey, "").nonEmpty) {
+ Some(assumedRolePolicyReason(bucket, S3AssumedRolePolicyKey))
+ } else {
+ None
+ }
+ } catch {
+ case e @ (_: IOException | _: RuntimeException) =>
+ Some(unverifiableValueReason(bucket, S3AssumedRolePolicyKey, e))
+ }
+ }
+ }
+ }
+
+ /** Hadoop's S3A SSL switch; `S3AFileSystem` reads it via
`Configuration#getBoolean`. */
+ private val S3SslEnabledKey = "fs.s3a.connection.ssl.enabled"
+ private val S3EndpointKey = "fs.s3a.endpoint"
+
+ /**
+ * Hadoop's assumed-role STS endpoint keys; `AssumedRoleCredentialProvider`
sends AssumeRole to
+ * the configured endpoint, while native builds its provider with SDK
defaults.
+ */
+ private val S3StsEndpointKeys =
+ Seq("fs.s3a.assumed.role.sts.endpoint",
"fs.s3a.assumed.role.sts.endpoint.region")
+
+ private def insecureEndpointReason(bucket: String): String =
+ s"Native scan does not support a scheme-less $S3EndpointKey with " +
+ s"$S3SslEnabledKey=false for $bucket (Hadoop addresses that endpoint
over http://, " +
+ "while the native S3 client always assumes https://, so a claimed scan
would fail at " +
+ "execution where Spark reads fine)"
+
+ private def stsEndpointReason(bucket: String, key: String): String =
+ s"Native scan does not support $key configured for $bucket (Hadoop sends
its " +
+ "AssumeRole request to the configured STS endpoint, while the native S3
client's " +
+ "assumed-role provider uses the SDK defaults, so the two sides would
authenticate " +
+ "against different endpoints)"
+
+ /**
+ * First reason any bucket among `uris` configures an endpoint native would
address differently
+ * from Hadoop: a scheme-less `fs.s3a.endpoint` with SSL disabled, or an
assumed-role STS
+ * endpoint. Resolved like the keys' real consumers (plain reads on the
propagated conf,
+ * mirroring [[proxyGateReason]]); the STS keys decline whenever set, since
dead config can
+ * become live through a provider-chain change native never re-validates.
+ */
+ private[comet] def hadoopOnlyEndpointGateReason(
+ hadoopConf: Configuration,
+ uris: Seq[URI],
+ propagatedConfCache: MutableMap[String, Configuration] =
MutableMap.empty)
+ : Option[String] = {
+ val buckets = uris.flatMap(s3Bucket).distinct
+ buckets.foldLeft(Option.empty[String]) { (declined, bucket) =>
+ if (declined.isDefined) {
+ declined
+ } else {
+ try {
+ val propagatedConf =
+ propagatedConfCache.getOrElseUpdate(
+ bucket,
+ propagateBucketOptions(hadoopConf, bucket))
+ val endpoint = propagatedConf.getTrimmed(S3EndpointKey, "")
+ val schemeless = endpoint.nonEmpty && !endpoint.contains("://")
+ if (schemeless && !propagatedConf.getBoolean(S3SslEnabledKey, true))
{
+ Some(insecureEndpointReason(bucket))
+ } else {
+ S3StsEndpointKeys
+ .find(key => propagatedConf.getTrimmed(key, "").nonEmpty)
Review Comment:
The SPI exemption added to `assumedRolePolicyGateReason` and
`providerClassGateReason` (through `cometProviderClassConfigured`) seems to
apply equally to the STS endpoint keys, but this check still declines them.
When `fs.s3a.comet.credential.provider.class` resolves for a bucket,
`create_store` in `s3.rs` takes the `CometS3CredentialBridge` branch and never
calls `build_credential_provider`, the native assumed-role provider that
ignores `fs.s3a.assumed.role.sts.endpoint`.
`HadoopS3ACredentialProviderAdapter` builds Hadoop's own provider chain over
the forwarded `fs.s3a.*` keys, so AssumeRole should go to the configured
endpoint. With this change a regular Parquet scan using that setup plus an STS
endpoint key falls back to Spark, where on main it runs natively. Could the STS
check skip buckets where `cometProviderClassConfigured` is true, keeping the
scheme-less endpoint check, which applies to the store itself? A test next to
`assumedRolePolicyGateReason admits a session policy
when the Comet SPI is set` would pin it. I read this from the sources and did
not run it.
--
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]
