diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index b9754029911..a24cc12a8ce 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -462,7 +462,7 @@ For each `abfs` or `abfss` path, the Spark driver asks the hadoop-azure library The native scan receives the resolved values and builds the `object_store` client from them. It reads no `fs.azure.*` key of its own. A scan whose files span more than one ABFS container or account, or that mixes `abfs`/`abfss` paths with other schemes, falls back to Spark's own scan. -hadoop-azure must be on the driver classpath. It already is whenever Spark can list the path. When it is missing, the scan fails with an error that says so. +hadoop-azure must be on the driver classpath. It already is whenever Spark can list the path. When the driver cannot load it, or its `AbfsConfiguration` does not match the versions above (a vendor fork, for example), the scan falls back to Spark's own scan. #### Mechanisms @@ -479,15 +479,15 @@ hadoop-azure must be on the driver classpath. It already is whenever Spark can l | `WorkloadIdentityTokenProvider` with a `fs.azure.account.oauth2.client.assertion.provider.type` (hadoop-azure 3.5.0 and later) | Not supported | | | `UserboundSASWithOAuth` (hadoop-azure 3.5.0 and later) | Not supported | | -Custom provider classes hand each token request to Java code the native scan cannot call, and the refresh token and user password flows have no `object_store` counterpart. When Hadoop selects one of these, the scan fails with an error naming the auth type and, where one is configured, the class. Configure a supported mechanism for that account, or keep the path on Spark's own scan. Values arrive as Hadoop resolved them, trimmed and defaulted where the ABFS driver trims and defaults them, so the store sees what the driver would have used. +Custom provider classes hand each token request to Java code the native scan cannot call, and the refresh token and user password flows have no `object_store` counterpart. When Hadoop selects one of these, the scan falls back to Spark's own scan, which authenticates through hadoop-azure, and the fallback reason names the auth type and, for SAS, OAuth and Workload Identity, the provider class (`Custom` names only the auth type). Configure a supported mechanism for that account to read it natively. Values arrive as Hadoop resolved them, trimmed and defaulted where the ABFS driver trims and defaults them, so the store sees what the driver would have used. -Client credentials and Workload Identity token requests go over HTTPS only. The managed identity request goes to the IMDS endpoint over HTTP, as Hadoop's does. For client credentials, Hadoop posts to `fs.azure.account.oauth2.client.endpoint` as given, while `object_store` posts to `//oauth2/v2.0/token`. Comet takes the tenant from the path segment before the last `oauth2` and everything before that segment as the authority host, so a v1 `/oauth2/token` endpoint and a proxy with a path prefix both resolve. The token request path is always the v2.0 one. An `http://` endpoint fails. +Client credentials and Workload Identity token requests go over HTTPS only. The managed identity request goes to the IMDS endpoint over HTTP, as Hadoop's does. For client credentials, Hadoop posts to `fs.azure.account.oauth2.client.endpoint` as given, while `object_store` posts to `//oauth2/v2.0/token`. Comet takes the tenant from the path segment before the last `oauth2` and everything before that segment as the authority host, so a v1 `/oauth2/token` endpoint and a proxy with a path prefix both resolve. The token request path is always the v2.0 one. An `http://` endpoint, or one with no tenant segment, makes the scan fall back to Spark's own scan. So does an `http://` Workload Identity authority (`fs.azure.account.oauth2.msi.authority`). The driver decides this by asking the native store whether it can be built from the resolved values, so the scan falls back exactly when the native store would reject them. A native write to such an account fails with an error naming the key. ### Environment variables The `AZURE_*` variables (`AZURE_CLIENT_ID`, `AZURE_TENANT_ID`, `AZURE_FEDERATED_TOKEN_FILE`, `AZURE_STORAGE_ACCOUNT_KEY`, `AZURE_USE_AZURE_CLI`, and the others `object_store` reads) and `IDENTITY_ENDPOINT` are read only when Hadoop configures no authentication for the account at all: none of `fs.azure.account.auth.type`, `fs.azure.account.key`, `fs.azure.account.keyprovider`, `fs.azure.shellkeyprovider.script`, `fs.azure.account.oauth.provider.type`, `fs.azure.sas.token.provider.type`, `fs.azure.sas.fixed.token` or any `fs.azure.account.oauth2.*` key resolves for it. That case is what lets AKS Workload Identity work with no Hadoop configuration for tables whose files Spark does not list through Hadoop, such as Iceberg tables. A plain Parquet path still needs Hadoop configured for Spark's own listing. -Once Hadoop names anything, only the resolved values reach the store. A resolution error (a missing mandatory key, an invalid account key, an unknown provider class) fails the scan with Hadoop's exception. Messages that could quote a credential, such as the key provider's, are kept in the Spark driver log and the scan error names the exception class. The environment is never a fallback. One read happens inside `object_store` rather than Comet: under a resolved managed identity, its IMDS provider picks up `IDENTITY_HEADER` when it fetches a token. +Once Hadoop names anything, only the resolved values reach the store. A resolution error (a missing mandatory key, an invalid account key, an unknown provider class) makes the scan fall back to Spark's own scan, which reports the error in hadoop-azure's own words if it cannot authenticate either. The fallback reason carries Hadoop's exception. Messages that could quote a credential, such as the key provider's, are kept in the Spark driver log and the reason names the exception class. The environment is never a fallback. One read happens inside `object_store` rather than Comet: under a resolved managed identity, its IMDS provider picks up `IDENTITY_HEADER` when it fetches a token. ### Credentials in transit @@ -557,7 +557,7 @@ Comet 1.1 started from the environment and overlaid the Hadoop keys it found by - Transport variables such as `AZURE_STORAGE_ENDPOINT` and `AZURE_ALLOW_HTTP` no longer apply beside a Hadoop mechanism. They are read only when Hadoop configures none. - The WASB key `fs.azure.sas..`, which the ABFS driver never read, is no longer read by the native scan either. Use `fs.azure.sas.fixed.token`, which the native scan now reads. - A lower-case `oauth` auth type, or OAuth keys without `fs.azure.account.auth.type`, fail the way they fail for Spark. In the second case Hadoop defaults to `SharedKey` and the key provider fails. On hadoop-azure 3.4.1 and later the driver log names the missing key. 3.3.4 reports only `Failure to initialize configuration`. -- A configured mechanism the native scan cannot build fails with an error instead of silently reading the environment. +- A configured mechanism the native scan cannot build falls back to Spark's own scan instead of silently reading the environment. So does a driver whose hadoop-azure Comet cannot load. ### Azure Limitations diff --git a/native/core/src/lib.rs b/native/core/src/lib.rs index 4675a59d576..fd1fef05041 100644 --- a/native/core/src/lib.rs +++ b/native/core/src/lib.rs @@ -30,7 +30,7 @@ extern crate core; extern crate datafusion_comet_jni_bridge; use jni::{ - objects::{JClass, JString}, + objects::{JClass, JObjectArray, JString}, sys::{jboolean, jstring, JNI_FALSE}, EnvUnowned, }; @@ -41,6 +41,7 @@ use log4rs::{ encode::pattern::PatternEncoder, Config, }; +use std::collections::HashMap; // Re-export from jvm-bridge crate for internal use pub use datafusion_comet_jni_bridge::errors; @@ -319,6 +320,35 @@ pub extern "system" fn Java_org_apache_comet_NativeBase_icebergStorageSchemes( }) } +/// JNI: why the native scan cannot build its ABFS store for `url` from the object store options +/// `keys`/`values`, or null when it can. The driver asks this when it plans a scan whose ABFS +/// authentication resolved, so a store native would reject falls back to Spark instead of every +/// task failing. See `azure::store_error`. +#[no_mangle] +pub extern "system" fn Java_org_apache_comet_NativeBase_abfsStoreError( + env: EnvUnowned, + _: JClass, + url: JString, + keys: JObjectArray, + values: JObjectArray, +) -> jstring { + try_unwrap_or_throw(&env, |env| { + let url: String = url.try_to_string(env)?; + let mut configs = HashMap::new(); + for i in 0..keys.len(env)? { + let key = keys.get_element(env, i)?; + let key = unsafe { JString::from_raw(&*env, key.into_raw()) }.try_to_string(env)?; + let value = values.get_element(env, i)?; + let value = unsafe { JString::from_raw(&*env, value.into_raw()) }.try_to_string(env)?; + configs.insert(key, value); + } + match parquet::objectstore::azure::store_error(&url, &configs) { + Some(message) => Ok(env.new_string(message)?.into_raw()), + None => Ok(std::ptr::null_mut()), + } + }) +} + // Creates a default log4rs config, which logs to console with log level. fn default_logger_config(log_level: &str) -> CometResult { let console_append = ConsoleAppender::builder() diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs index 4b49cc07eb3..8b18ac70250 100644 --- a/native/core/src/parquet/objectstore/azure.rs +++ b/native/core/src/parquet/objectstore/azure.rs @@ -37,6 +37,11 @@ //! - `error`: an error carrying the driver's exception class and message. //! - missing or anything else: an error, so a plan that did not run the resolver cannot pass. //! +//! Scans never reach any of these errors. The driver handles `declined` and `error` itself and, +//! for `resolved`, asks [`store_error`] (through `NativeBase.abfsStoreError`) whether +//! [`create_store`] can build the store the scan would build. A scan that would fail here is +//! planned on Spark instead. Native writes still fail with these errors. +//! //! `comet.azure.auth.type` is Hadoop's `AuthType` name, `comet.azure.oauth.provider.class` the //! OAuth provider class, `comet.azure.sas.provider.class` a custom SAS provider and //! `comet.azure.oauth.assertion.provider.class` a Workload Identity client assertion provider. @@ -62,7 +67,9 @@ //! the client id and tenant to its request only when they are non-empty. The MSI tenant and //! authority are accepted but unused: object_store's IMDS provider reads only the client id and //! endpoint. `RefreshTokenBasedTokenProvider` and `UserPasswordTokenProvider` have no -//! `object_store` counterpart and fail as unsupported. A value a mechanism needs but that is +//! `object_store` counterpart and fail as unsupported. The client credentials endpoint and the +//! Workload Identity authority must be `https` URLs: the store is built without `allow_http`, so +//! an `http` token request would fail on every attempt. A value a mechanism needs but that is //! absent (or, for the SAS token and the client credentials endpoint, unusable) is an error //! naming the key; other `fs.azure.*` keys are ignored. Values are applied as forwarded (Hadoop //! already trimmed and defaulted them), except that the trailing slash Hadoop appends to @@ -129,6 +136,21 @@ pub fn create_store( Ok((Box::new(store), path)) } +/// Why [`create_store`] cannot build a store for `url` from `configs`, or `None` when it can. +/// The driver asks this at planning time so that a scan the store would reject falls back to +/// Spark, from the same code the scan runs instead of a copy of its rules. Building makes no +/// request, and the message names keys, never their values. +pub fn store_error(url: &str, configs: &HashMap) -> Option { + let url = match Url::parse(url) { + Ok(url) => url, + Err(e) => return Some(format!("invalid ABFS URL: {e}")), + }; + create_store(&url, configs).err().map(|e| match e { + object_store::Error::Generic { source, .. } => source.to_string(), + other => other.to_string(), + }) +} + /// The builder the markers in `configs` call for. Only `none` touches the environment. fn builder_for( url: &Url, @@ -232,6 +254,7 @@ fn client_creds_settings( before `oauth2`" )) })?; + require_https(HADOOP_OAUTH_CLIENT_ENDPOINT, &host)?; Ok(vec![ ( AzureConfigKey::ClientId, @@ -274,6 +297,14 @@ fn msi_settings(configs: &HashMap) -> Result, ) -> Result { + let authority = required(configs, HADOOP_MSI_AUTHORITY)?; + let has_host = Url::parse(&authority).is_ok_and(|url| url.has_host()); + if !has_host { + return Err(config_error(format!( + "`{HADOOP_MSI_AUTHORITY}` has no `scheme://host` origin" + ))); + } + require_https(HADOOP_MSI_AUTHORITY, &authority)?; Ok(vec![ ( AzureConfigKey::ClientId, @@ -287,13 +318,23 @@ fn workload_identity_settings( AzureConfigKey::FederatedTokenFile, required(configs, HADOOP_WI_TOKEN_FILE)?, ), - ( - AzureConfigKey::AuthorityHost, - authority_host(&required(configs, HADOOP_MSI_AUTHORITY)?), - ), + (AzureConfigKey::AuthorityHost, authority_host(&authority)), ]) } +/// The store is built without `allow_http`, so a token request to an `http` URL fails on every +/// attempt. Fail at build time instead, naming the key. `url` must already parse. +fn require_https(key: &str, url: &str) -> Result<(), object_store::Error> { + if Url::parse(url).is_ok_and(|url| url.scheme() == "https") { + Ok(()) + } else { + Err(config_error(format!( + "`{key}` is not an `https` URL; the native ABFS store sends token requests over \ + HTTPS only" + ))) + } +} + fn required(configs: &HashMap, key: &str) -> Result { configs.get(key).cloned().ok_or_else(|| { config_error(format!( @@ -529,6 +570,33 @@ mod tests { assert!(err.contains("Scheme of URL is not Azure"), "{err}"); } + #[test] + fn store_error_answers_whether_create_store_builds() { + let shared_key = with(shared_key(), HADOOP_KEY, KEY); + assert_eq!(store_error(URL, &shared_key), None); + assert_eq!(store_error(URL, &client_creds()), None); + let http = with( + client_creds(), + HADOOP_OAUTH_CLIENT_ENDPOINT, + "http://login.microsoftonline.com/hadoop-tenant/oauth2/token", + ); + let err = store_error(URL, &http).expect("an error"); + assert!( + err.starts_with(&format!( + "`{HADOOP_OAUTH_CLIENT_ENDPOINT}` is not an `https` URL" + )), + "{err}" + ); + let declined = configs(&[ + (MARKER_RESOLUTION, "declined"), + (MARKER_AUTH_TYPE, "Custom"), + ]); + let err = store_error(URL, &declined).expect("an error"); + assert!(err.contains("auth type Custom"), "{err}"); + let err = store_error("not a url", &shared_key).expect("an error"); + assert!(err.starts_with("invalid ABFS URL"), "{err}"); + } + #[test] fn client_creds_take_tenant_and_authority_host_from_the_endpoint_alone() { assert_exactly( @@ -563,9 +631,11 @@ mod tests { ); // An endpoint without a tenant segment or without a `scheme://host` origin cannot // fall through to another credential or to the default authority host. + // Nor can an `http` endpoint, which the store would refuse on every token request. for endpoint in [ "https://auth-proxy.example/", "file:///hadoop-tenant/oauth2/token", + "http://login.microsoftonline.com/hadoop-tenant/oauth2/token", ] { let err = err(&with( client_creds(), @@ -694,6 +764,15 @@ mod tests { "{authority}" ); } + // The token request goes to the authority over a client that refuses `http`. + for authority in [ + "http://login.microsoftonline.com/", + "login.microsoftonline.com", + "file:///", + ] { + let err = err(&with(workload_identity(), HADOOP_MSI_AUTHORITY, authority)); + assert!(err.contains(HADOOP_MSI_AUTHORITY), "{authority}: {err}"); + } } #[test] diff --git a/spark/src/main/java/org/apache/comet/NativeBase.java b/spark/src/main/java/org/apache/comet/NativeBase.java index d09c92c74b1..b9349057c99 100644 --- a/spark/src/main/java/org/apache/comet/NativeBase.java +++ b/spark/src/main/java/org/apache/comet/NativeBase.java @@ -401,4 +401,17 @@ public static void releaseNative() throws Throwable { * @return the supported schemes joined by commas, e.g. "file,memory,gs,s3,s3a" */ public static native String icebergStorageSchemes(boolean forWrite); + + /** + * Why the native scan cannot build its ABFS object store for {@code url} from the given object + * store options, or null when it can. This runs the store construction the scan itself runs, so + * the planner falls back on exactly the configurations native rejects rather than a copy of its + * rules. Building the store makes no request, and the message names option keys, never values. + * + * @param url an {@code abfs} or {@code abfss} URL naming the container and account + * @param keys the option keys, as produced by {@code AbfsAuthResolver.toOptions} + * @param values the option values, in the order of {@code keys} + * @return the reason the store cannot be built, or null + */ + public static native String abfsStoreError(String url, String[] keys, String[] values); } diff --git a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala index a7938ce79a8..872a67d3f24 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala @@ -29,6 +29,7 @@ import org.apache.commons.lang3.StringUtils import org.apache.hadoop.conf.Configuration import org.apache.spark.internal.Logging +import org.apache.comet.NativeBase import org.apache.comet.objectstore.AbfsReflection.Handles /** @@ -41,7 +42,9 @@ import org.apache.comet.objectstore.AbfsReflection.Handles * (container-, account- or globally-scoped keys, credential providers, `${...}` references). The * resolver forwards the resolved values under the global key names plus `comet.azure.*` markers, * never a key for another account, and never the environment. Anything Hadoop throws becomes an - * error marker: the native scan fails with it instead of trying a lookup of its own. + * error marker instead of a lookup of its own. A scan whose resolution the native store cannot + * use falls back to Spark (see [[nativeFallbackReason]]); the markers still carry the outcome to + * native for the other callers, which fail with it. * * `AbfsAuthParitySuite` pins the outcome of every configuration shape per hadoop-azure version * (`spark/src/test/resources/abfs-auth-parity/expected-.json`) and cross-checks it @@ -467,6 +470,57 @@ private[comet] object AbfsAuthResolver extends Logging { Markers.ERROR_MESSAGE -> message) } + /** + * Why the native store cannot read `uri` under `outcome`, or None when it can: a mechanism + * Hadoop named that runs Java code per request, a resolution that failed, or resolved values + * the native store rejects. All are known on the driver, so the scan falls back to Spark's own + * reader, which authenticates through hadoop-azure, instead of planning natively and failing + * every task. For `Resolved`, native itself answers (see [[nativeStoreError]]), so the rules + * live in one place. + */ + def nativeFallbackReason(uri: URI, outcome: Outcome): Option[String] = { + // The container and host only: a password in the user info stays out of the reason. + val authority = Option(uri.getRawAuthority).getOrElse("") + val at = authority.lastIndexOf('@') + val safeAuthority = + if (at < 0) authority + else authority.substring(0, at).takeWhile(_ != ':') + authority.substring(at) + val location = s"${uri.getScheme}://$safeAuthority" + outcome match { + case Declined(authType, providerClass, assertionProviderClass) => + val classes = providerClass.map(c => s" with provider $c").getOrElse("") + + assertionProviderClass.map(c => s" and client assertion provider $c").getOrElse("") + Some( + s"Native scan does not support ABFS authentication auth type $authType$classes " + + s"for $location") + case resolved: Resolved => + // The message names option keys, never their values. + nativeStoreError(s"$location/", toOptions(resolved)).map(message => + s"Native scan cannot use the ABFS authentication for $location: $message") + case Failed(_, message) => + // Some of Hadoop's own messages quote the whole URI, user info included. + val safeMessage = + if (authority.isEmpty) message else message.replace(authority, safeAuthority) + Some(s"Hadoop ABFS authentication failed on the driver for $location: $safeMessage") + case Unconfigured => None + } + } + + /** + * Why native cannot build the store for `url` from `options`, from the same code the scan runs. + * When native cannot be asked (the library is not loaded), this assumes it can: such a build + * cannot run the native scan anyway. + */ + private def nativeStoreError(url: String, options: Map[String, String]): Option[String] = { + val (keys, values) = options.toArray.unzip + try Option(NativeBase.abfsStoreError(url, keys, values)) + catch { + case e @ (NonFatal(_) | _: LinkageError) => + logDebug(s"Could not ask native whether it can build the ABFS store for $url", e) + None + } + } + private def providerMarker(authType: String): String = if (authType == AuthTypes.SAS) Markers.SAS_PROVIDER_CLASS else Markers.OAUTH_PROVIDER_CLASS diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index 0cdf8183bfe..b6f63ac77ef 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -341,19 +341,8 @@ object NativeConfig { val scheme = Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("file") import scala.jdk.CollectionConverters._ - val options = scala.collection.mutable.Map[String, String]() - - // Scheme lists consumed on the native side ride along in the options map (both empty by - // default). - val libhdfsSchemes = hadoopConf.get(COMET_LIBHDFS_SCHEMES_KEY) - if (StringUtils.isNotBlank(libhdfsSchemes)) { - options(COMET_LIBHDFS_SCHEMES_KEY) = libhdfsSchemes - } - val s3CompliantRaw = hadoopConf.get(COMET_S3_COMPLIANT_SCHEMES_KEY) - if (StringUtils.isNotBlank(s3CompliantRaw)) { - options(COMET_S3_COMPLIANT_SCHEMES_KEY) = s3CompliantRaw - } - val s3CompliantSchemes = parseSchemeSet(s3CompliantRaw) + val options = scala.collection.mutable.Map[String, String]() ++= schemeListOptions(hadoopConf) + val s3CompliantSchemes = resolveS3CompliantSchemes(hadoopConf) // ABFS: Hadoop resolves the mechanism and values for this URI's account on the driver, and // only those travel to native (never another account's keys, never the whole namespace). @@ -399,6 +388,12 @@ object NativeConfig { (options.toMap, aliasOptions) } + /** The scheme lists consumed on the native side; both are empty by default. */ + private def schemeListOptions(hadoopConf: Configuration): Map[String, String] = + Seq(COMET_LIBHDFS_SCHEMES_KEY, COMET_S3_COMPLIANT_SCHEMES_KEY).flatMap { key => + Option(hadoopConf.get(key)).filter(StringUtils.isNotBlank).map(key -> _) + }.toMap + /** * The object store options for a scan over `uris`, or the reason the native scan cannot read * them. A scan without ABFS files forwards the options of every scheme it reads (see @@ -406,6 +401,8 @@ 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. + * A resolution the native store cannot use (see [[AbfsAuthResolver.nativeFallbackReason]]) + * declines the scan too, so that Spark reads it instead of every task failing. */ def extractScanObjectStoreOptions( hadoopConf: Configuration, @@ -430,7 +427,11 @@ object NativeConfig { } } if (firstIsAbfs) { - Right(extractObjectStoreOptions(hadoopConf, first)) + val outcome = AbfsAuthResolver.resolve(hadoopConf, first) + AbfsAuthResolver.nativeFallbackReason(first, outcome) match { + case Some(reason) => Left(reason) + case None => Right(schemeListOptions(hadoopConf) ++ AbfsAuthResolver.toOptions(outcome)) + } } else { Right(extractObjectStoreOptions(hadoopConf, uris)) } diff --git a/spark/src/test/scala/org/apache/comet/objectstore/AbfsScanFallbackSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/AbfsScanFallbackSuite.scala index 016491d8af8..5386697de1b 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/AbfsScanFallbackSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/AbfsScanFallbackSuite.scala @@ -112,6 +112,41 @@ class AbfsScanFallbackSuite extends CometTestBase { } } + private def abfsKeys(pairs: (String, String)*): Seq[(String, String)] = + pairs.map { case (key, value) => s"$key.accta.dfs.core.windows.net" -> value } + + test("native Parquet scan with a Custom ABFS auth type falls back to Spark") { + writeParquet("parquet-custom", 0) + withSQLConf(abfsKeys("fs.azure.account.auth.type" -> "Custom"): _*) { + val df = spark.read.parquet(abfss("data", "accta", "parquet-custom")) + checkSparkAnswerAndFallbackReason(df, "auth type Custom") + } + } + + test("native Parquet scan with an unknown ABFS auth type falls back to Spark") { + writeParquet("parquet-unknown-auth", 0) + withSQLConf(abfsKeys("fs.azure.account.auth.type" -> "NotAnAuthType"): _*) { + val df = spark.read.parquet(abfss("data", "accta", "parquet-unknown-auth")) + checkSparkAnswerAndFallbackReason(df, "failed on the driver") + } + } + + test("native CSV scan with a Custom ABFS auth type falls back to Spark") { + writeCsv("csv-custom", 0) + withSQLConf(csvConfs ++ abfsKeys("fs.azure.account.auth.type" -> "Custom"): _*) { + val df = readCsv(abfss("data", "accta", "csv-custom")) + checkSparkAnswerAndFallbackReason(df, "auth type Custom") + } + } + + test("native CSV scan with an unknown ABFS auth type falls back to Spark") { + writeCsv("csv-unknown-auth", 0) + withSQLConf(csvConfs ++ abfsKeys("fs.azure.account.auth.type" -> "NotAnAuthType"): _*) { + val df = readCsv(abfss("data", "accta", "csv-unknown-auth")) + checkSparkAnswerAndFallbackReason(df, "failed on the driver") + } + } + test("native Parquet scan over two directories of one container stays native") { writeParquet("parquet-e", 0) writeParquet("parquet-f", 100) diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index 3863f66105a..8c4ad9256ef 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -27,6 +27,7 @@ import org.scalatest.matchers.should.Matchers import org.apache.hadoop.conf.Configuration import org.apache.comet.CometConf.{COMET_LIBHDFS_SCHEMES_KEY, COMET_S3_COMPLIANT_SCHEMES_KEY} +import org.apache.comet.objectstore.AbfsAuthResolver.{Keys, Markers, Providers} class NativeConfigSuite extends AnyFunSuite with Matchers { @@ -755,6 +756,235 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { } } + private def abfsScanConf(pairs: (String, String)*): Configuration = { + val hadoopConf = new Configuration() + pairs.foreach { case (key, value) => + hadoopConf.set(s"$key.accta.dfs.core.windows.net", value) + } + hadoopConf + } + + private def assertFallsBack(hadoopConf: Configuration, expected: String*): Unit = { + val uris = Seq(abfsFile("data", "accta", "a.parquet"), abfsFile("data", "accta", "b.parquet")) + NativeConfig.extractScanObjectStoreOptions(hadoopConf, uris) match { + case Left(reason) => + reason should include("abfss://data@accta.dfs.core.windows.net") + expected.foreach(text => reason should include(text)) + case Right(options) => fail(s"expected a fallback, got $options") + } + } + + test("extractScanObjectStoreOptions - a Custom ABFS auth type falls back to Spark") { + assertFallsBack( + abfsScanConf( + Keys.AUTH_TYPE -> "Custom", + Keys.OAUTH_PROVIDER_TYPE -> classOf[NeverInstantiatedAccessTokenProvider].getName), + "auth type Custom") + } + + test("extractScanObjectStoreOptions - an ABFS fallback reason leaves out a URL password") { + val hadoopConf = abfsScanConf(Keys.AUTH_TYPE -> "Custom") + val uri = new URI("abfss://data:url-password@accta.dfs.core.windows.net/path/a.parquet") + NativeConfig.extractScanObjectStoreOptions(hadoopConf, Seq(uri)) match { + case Left(reason) => + reason should include("abfss://data@accta.dfs.core.windows.net") + reason should not include "url-password" + case Right(options) => fail(s"expected a fallback, got $options") + } + } + + test("extractScanObjectStoreOptions - a URL password stays out of a Hadoop message") { + // Hadoop reads `data:url-password` as the container and rejects the short account name with + // a message that quotes the whole URI. + val hadoopConf = new Configuration() + hadoopConf.set("fs.azure.account.key.accta", "a2V5LWE=") + val uri = new URI("abfss://data:url-password@accta/path/a.parquet") + NativeConfig.extractScanObjectStoreOptions(hadoopConf, Seq(uri)) match { + case Left(reason) => + reason should include("account name is not fully qualified") + reason should include("abfss://data@accta") + reason should not include "url-password" + case Right(options) => fail(s"expected a fallback, got $options") + } + } + + test("nativeFallbackReason - names the classes of a declined Workload Identity and SAS") { + val uri = abfsFile("data", "accta", "a.parquet") + val assertion = AbfsAuthResolver.Declined( + "OAuth", + Some(Providers.WORKLOAD_IDENTITY), + Some("com.example.Assertion")) + val reason = AbfsAuthResolver.nativeFallbackReason(uri, assertion).get + reason should include("auth type OAuth") + reason should include(s"with provider ${Providers.WORKLOAD_IDENTITY}") + reason should include("and client assertion provider com.example.Assertion") + + val userbound = AbfsAuthResolver.Declined("UserboundSASWithOAuth", None) + AbfsAuthResolver.nativeFallbackReason(uri, userbound).get should include( + "auth type UserboundSASWithOAuth") + } + + test("extractScanObjectStoreOptions - a custom SAS token provider falls back to Spark") { + val providerClass = classOf[NeverInstantiatedSasTokenProvider].getName + assertFallsBack( + abfsScanConf(Keys.AUTH_TYPE -> "SAS", Keys.SAS_TOKEN_PROVIDER_TYPE -> providerClass), + "auth type SAS", + providerClass) + } + + test("extractScanObjectStoreOptions - RefreshToken and UserPassword OAuth fall back to Spark") { + assertFallsBack( + abfsScanConf( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.REFRESH_TOKEN, + Keys.REFRESH_TOKEN -> "refresh-token-value", + Keys.CLIENT_ID -> "client-123"), + Providers.REFRESH_TOKEN) + assertFallsBack( + abfsScanConf( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.USER_PASSWORD, + Keys.CLIENT_ENDPOINT -> "https://login.microsoftonline.com/tenant/oauth2/token", + Keys.USER_NAME -> "user@example.com", + Keys.USER_PASSWORD -> "user-password-value"), + Providers.USER_PASSWORD) + } + + private def clientCredsConf(endpoint: String): Configuration = + abfsScanConf( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.CLIENT_CREDS, + Keys.CLIENT_ENDPOINT -> endpoint, + Keys.CLIENT_ID -> "client-123", + Keys.CLIENT_SECRET -> "client-secret-value") + + test("extractScanObjectStoreOptions - a client credentials endpoint follows native") { + // Native answers for the endpoint: `azure.rs` builds no store for an `http` endpoint or one + // without a tenant segment, and resolves the URL parser's quirks rather than rejecting them. + Seq( + "http://login.microsoftonline.com/tenant/oauth2/token" -> "is not an `https` URL", + "https://login.microsoftonline.com/" -> "no tenant segment", + "https://login.microsoftonline.com" -> "no tenant segment").foreach { + case (endpoint, message) => + withClue(s"$endpoint: ") { + assertFallsBack(clientCredsConf(endpoint), Keys.CLIENT_ENDPOINT, message) + } + } + val uris = Seq(abfsFile("data", "accta", "a.parquet")) + Seq( + "https://auth-proxy.example/aad/t/oauth2/v2.0/token", + "https://h/a/%2e%2e/t/oauth2/token", + "https://h/a\\b/t/oauth2/token").foreach { endpoint => + withClue(s"$endpoint: ") { + NativeConfig.extractScanObjectStoreOptions(clientCredsConf(endpoint), uris) match { + case Right(options) => options(Markers.RESOLUTION) shouldBe Markers.RESOLVED + case Left(reason) => fail(s"expected native options, got a fallback: $reason") + } + } + } + } + + test("extractScanObjectStoreOptions - a client credentials endpoint password stays out") { + val hadoopConf = clientCredsConf("http://user:endpoint-password@login.example/t/oauth2/token") + NativeConfig.extractScanObjectStoreOptions( + hadoopConf, + Seq(abfsFile("data", "accta", "a.parquet"))) match { + case Left(reason) => reason should not include "endpoint-password" + case Right(options) => fail(s"expected a fallback, got $options") + } + } + + test("extractScanObjectStoreOptions - an http Workload Identity authority falls back") { + val handles = AbfsReflection.handles(getClass.getClassLoader).get + assume(handles.hasFixedSasTokenProvider, "Workload Identity needs hadoop-azure 3.4.1+") + def conf(authority: String): Configuration = + abfsScanConf( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.WORKLOAD_IDENTITY, + Keys.MSI_TENANT -> "tenant-abc", + Keys.CLIENT_ID -> "client-123", + Keys.MSI_AUTHORITY -> authority) + assertFallsBack( + conf("http://user:authority-password@login.example/"), + Keys.MSI_AUTHORITY, + "is not an `https` URL") + val uris = Seq(abfsFile("data", "accta", "a.parquet")) + NativeConfig.extractScanObjectStoreOptions( + conf("http://user:authority-password@login.example/"), + uris) match { + case Left(reason) => reason should not include "authority-password" + case Right(options) => fail(s"expected a fallback, got $options") + } + NativeConfig.extractScanObjectStoreOptions(conf("https://login.example/aad/"), uris) match { + case Right(options) => options(Markers.RESOLUTION) shouldBe Markers.RESOLVED + case Left(reason) => fail(s"expected native options, got a fallback: $reason") + } + } + + test("extractScanObjectStoreOptions - a failed ABFS resolution falls back to Spark") { + assertFallsBack( + abfsScanConf(Keys.AUTH_TYPE -> "NotAnAuthType"), + "failed on the driver", + "No enum constant") + } + + test("extractScanObjectStoreOptions - hadoop-azure the driver cannot load falls back") { + // Also covers a configuration that sets no fs.azure.* key: the handles load first. + val hadoopConf = new Configuration() + hadoopConf.setClassLoader(new FlippingLoader(getClass.getClassLoader)) + assertFallsBack(hadoopConf, "hadoop-azure is not on the driver classpath") + } + + test("extractScanObjectStoreOptions - unconfigured and resolved ABFS stay native") { + val uris = Seq(abfsFile("data", "accta", "a.parquet")) + val hadoopConf = new Configuration() + hadoopConf.set(COMET_LIBHDFS_SCHEMES_KEY, "hdfs,viewfs") + NativeConfig.extractScanObjectStoreOptions(hadoopConf, uris) shouldBe Right( + Map(COMET_LIBHDFS_SCHEMES_KEY -> "hdfs,viewfs", Markers.RESOLUTION -> Markers.NONE)) + + hadoopConf.set("fs.azure.account.key.accta.dfs.core.windows.net", "a2V5LWE=") + val options = NativeConfig.extractScanObjectStoreOptions(hadoopConf, uris).toOption.get + options shouldBe NativeConfig.extractObjectStoreOptions(hadoopConf, uris.head) + options(Markers.RESOLUTION) shouldBe Markers.RESOLVED + } + + test("extractScanObjectStoreOptions - every supported OAuth and SAS mechanism stays native") { + val handles = AbfsReflection.handles(getClass.getClassLoader).get + val clientEndpoint = "https://login.microsoftonline.com/tenant/oauth2/token" + val supported = Seq( + "ClientCredsTokenProvider" -> Seq( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.CLIENT_CREDS, + Keys.CLIENT_ENDPOINT -> clientEndpoint, + Keys.CLIENT_ID -> "client-123", + Keys.CLIENT_SECRET -> "client-secret-value"), + "MsiTokenProvider" -> Seq( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.MSI, + Keys.MSI_TENANT -> "tenant-abc", + Keys.CLIENT_ID -> "client-123")) + val needsNewerHadoop = Seq( + "WorkloadIdentityTokenProvider" -> Seq( + Keys.AUTH_TYPE -> "OAuth", + Keys.OAUTH_PROVIDER_TYPE -> Providers.WORKLOAD_IDENTITY, + Keys.MSI_TENANT -> "tenant-abc", + Keys.CLIENT_ID -> "client-123"), + "fixed SAS token" -> Seq( + Keys.AUTH_TYPE -> "SAS", + Keys.SAS_FIXED_TOKEN -> "sv=2020-08-04&sig=fixed-signature")) + val cases = supported ++ (if (handles.hasFixedSasTokenProvider) needsNewerHadoop else Nil) + val uris = Seq(abfsFile("data", "accta", "a.parquet")) + cases.foreach { case (name, pairs) => + withClue(s"$name: ") { + val hadoopConf = abfsScanConf(pairs: _*) + NativeConfig.extractScanObjectStoreOptions(hadoopConf, uris) match { + case Right(options) => options(Markers.RESOLUTION) shouldBe Markers.RESOLVED + case Left(reason) => fail(s"expected native options, got a fallback: $reason") + } + } + } + } + test("extractScanObjectStoreOptions - non-abfs scans forward every scheme's options") { val hadoopConf = new Configuration() hadoopConf.set("fs.s3a.access.key", "s3-access-key")