fix: fall back to Spark when the driver cannot use the ABFS authentication

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 `<authority host>/<tenant>/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 `<authority host>/<tenant>/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.<container>.<account>`, 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
 

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,33 @@ 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<JString>,
+    values: JObjectArray<JString>,
+) -> 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)?.try_to_string(env)?;
+            let value = values.get_element(env, i)?.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<Config> {
     let console_append = ConsoleAppender::builder()

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<String, String>) -> Option<String> {
+    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<String, String>) -> Result<Settings, object_st
 fn workload_identity_settings(
     configs: &HashMap<String, String>,
 ) -> Result<Settings, object_store::Error> {
+    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<String, String>, key: &str) -> Result<String, object_store::Error> {
     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]

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);
 }