From 61fe07541e7e4cf076a84545fec1b42aab0835e5 Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Sun, 25 Oct 2026 09:20:51 -0600
Subject: [PATCH 1/6] fix: fall back to Spark when the driver cannot use the
 ABFS authentication

A native ABFS scan whose authentication Hadoop declines, resolves to a
RefreshToken or UserPassword OAuth provider, or fails to resolve on the
driver now falls back to Spark's own scan instead of planning natively
and failing every task.

Closes #7202
---
 docs/source/user-guide/latest/datasources.md  |  8 +-
 native/core/src/parquet/objectstore/azure.rs  |  3 +
 .../comet/objectstore/AbfsAuthResolver.scala  | 33 +++++++-
 .../comet/objectstore/NativeConfig.scala      | 29 +++----
 .../comet/objectstore/NativeConfigSuite.scala | 80 +++++++++++++++++++
 5 files changed, 134 insertions(+), 19 deletions(-)

diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index b9754029911..f43f06aa2d8 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,7 +479,7 @@ 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, where one is configured, the class. 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.
 
@@ -487,7 +487,7 @@ Client credentials and Workload Identity token requests go over HTTPS only. The
 
 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
 
diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs
index 4b49cc07eb3..2dbf6712150 100644
--- a/native/core/src/parquet/objectstore/azure.rs
+++ b/native/core/src/parquet/objectstore/azure.rs
@@ -37,6 +37,9 @@
 //! - `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 the `declined` and `error` arms, or the providers `resolved_settings`
+//! rejects: the driver plans those on Spark instead. Native writes still fail with them.
+//!
 //! `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.
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..25515b849c6 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
@@ -41,7 +41,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-<version>.json`) and cross-checks it
@@ -467,6 +469,35 @@ 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 built-in OAuth provider with no
+   * `object_store` counterpart, or a resolution that failed. 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.
+   */
+  def nativeFallbackReason(uri: URI, outcome: Outcome): Option[String] = {
+    def unsupported(mechanism: String): String =
+      s"Native scan does not support ABFS authentication $mechanism for " +
+        s"${uri.getScheme}://${uri.getRawAuthority}"
+    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(unsupported(s"auth type $authType$classes"))
+      case Resolved(
+            AuthTypes.OAUTH,
+            Some(c @ (Providers.REFRESH_TOKEN | Providers.USER_PASSWORD)),
+            _) =>
+        Some(unsupported(s"auth type ${AuthTypes.OAUTH} with provider $c"))
+      case Failed(_, message) =>
+        Some(
+          "Hadoop ABFS authentication failed on the driver for " +
+            s"${uri.getScheme}://${uri.getRawAuthority}: $message")
+      case _: Resolved | Unconfigured => 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..fbdffa35a42 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 = parseSchemeSet(hadoopConf.get(COMET_S3_COMPLIANT_SCHEMES_KEY))
 
     // 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/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
index 3863f66105a..daa80a8fb0a 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,85 @@ 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 - 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)
+  }
+
+  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 - non-abfs scans forward every scheme's options") {
     val hadoopConf = new Configuration()
     hadoopConf.set("fs.s3a.access.key", "s3-access-key")

From c33edf3da2b9bf4ddf8e8dd69350675938352820 Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Sun, 25 Oct 2026 09:31:08 -0600
Subject: [PATCH 2/6] fix: keep a URL password out of the ABFS fallback reason

---
 .../apache/comet/objectstore/AbfsAuthResolver.scala | 13 ++++++++-----
 .../comet/objectstore/NativeConfigSuite.scala       | 11 +++++++++++
 2 files changed, 19 insertions(+), 5 deletions(-)

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 25515b849c6..4ca59f50f03 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
@@ -477,9 +477,14 @@ private[comet] object AbfsAuthResolver extends Logging {
    * planning natively and failing every task.
    */
   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 location = s"${uri.getScheme}://" +
+      (if (at < 0) authority
+       else authority.substring(0, at).takeWhile(_ != ':') + authority.substring(at))
     def unsupported(mechanism: String): String =
-      s"Native scan does not support ABFS authentication $mechanism for " +
-        s"${uri.getScheme}://${uri.getRawAuthority}"
+      s"Native scan does not support ABFS authentication $mechanism for $location"
     outcome match {
       case Declined(authType, providerClass, assertionProviderClass) =>
         val classes = providerClass.map(c => s" with provider $c").getOrElse("") +
@@ -491,9 +496,7 @@ private[comet] object AbfsAuthResolver extends Logging {
             _) =>
         Some(unsupported(s"auth type ${AuthTypes.OAUTH} with provider $c"))
       case Failed(_, message) =>
-        Some(
-          "Hadoop ABFS authentication failed on the driver for " +
-            s"${uri.getScheme}://${uri.getRawAuthority}: $message")
+        Some(s"Hadoop ABFS authentication failed on the driver for $location: $message")
       case _: Resolved | Unconfigured => None
     }
   }
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 daa80a8fb0a..0c9941bec07 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
@@ -782,6 +782,17 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
       "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 custom SAS token provider falls back to Spark") {
     val providerClass = classOf[NeverInstantiatedSasTokenProvider].getName
     assertFallsBack(

From c2a7adb75f280b525da303c456911dcf6212babc Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Sun, 25 Oct 2026 09:59:05 -0600
Subject: [PATCH 3/6] fix: keep a URL password out of Hadoop messages in the
 ABFS fallback reason

Strip the password from the user info wherever a Failed message quotes it,
cover the supported OAuth and SAS mechanisms staying native, and add
end-to-end fallback tests for Parquet and CSV scans.
---
 docs/source/user-guide/latest/datasources.md  |  2 +-
 .../comet/objectstore/AbfsAuthResolver.scala  | 12 ++--
 .../objectstore/AbfsScanFallbackSuite.scala   | 35 ++++++++++
 .../comet/objectstore/NativeConfigSuite.scala | 68 +++++++++++++++++++
 4 files changed, 112 insertions(+), 5 deletions(-)

diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index f43f06aa2d8..796098c7740 100644
--- a/docs/source/user-guide/latest/datasources.md
+++ b/docs/source/user-guide/latest/datasources.md
@@ -479,7 +479,7 @@ 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 falls back to Spark's own scan, which authenticates through hadoop-azure, and the fallback reason names the auth type and, where one is configured, the class. 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.
+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.
 
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 4ca59f50f03..ef9601c4b3e 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
@@ -480,9 +480,10 @@ private[comet] object AbfsAuthResolver extends Logging {
     // 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 location = s"${uri.getScheme}://" +
-      (if (at < 0) authority
-       else authority.substring(0, at).takeWhile(_ != ':') + authority.substring(at))
+    val safeAuthority =
+      if (at < 0) authority
+      else authority.substring(0, at).takeWhile(_ != ':') + authority.substring(at)
+    val location = s"${uri.getScheme}://$safeAuthority"
     def unsupported(mechanism: String): String =
       s"Native scan does not support ABFS authentication $mechanism for $location"
     outcome match {
@@ -496,7 +497,10 @@ private[comet] object AbfsAuthResolver extends Logging {
             _) =>
         Some(unsupported(s"auth type ${AuthTypes.OAUTH} with provider $c"))
       case Failed(_, message) =>
-        Some(s"Hadoop ABFS authentication failed on the driver for $location: $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 _: Resolved | Unconfigured => None
     }
   }
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 0c9941bec07..0c4aed1b931 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
@@ -793,6 +793,37 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
     }
   }
 
+  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(
@@ -846,6 +877,43 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
     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")

From 8c7c35d26cb31ce60919fb27a4c9cd2f54612181 Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Mon, 26 Oct 2026 09:29:57 -0600
Subject: [PATCH 4/6] fix: fall back for a client credentials endpoint the
 native ABFS store rejects

nativeFallbackReason now declines a Resolved ClientCredsTokenProvider whose
endpoint is not https or has no tenant segment, which azure.rs rejects, so the
scan reads through Spark instead of failing every task. The reason leaves the
endpoint out because it can carry user info.

Also reuse resolveS3CompliantSchemes in objectStoreOptionsAndAliasOptions.
---
 docs/source/user-guide/latest/datasources.md  |  2 +-
 native/core/src/parquet/objectstore/azure.rs  |  5 +-
 .../comet/objectstore/AbfsAuthResolver.scala  | 31 +++++++++-
 .../comet/objectstore/NativeConfig.scala      |  2 +-
 .../comet/objectstore/NativeConfigSuite.scala | 62 +++++++++++++++++++
 5 files changed, 95 insertions(+), 7 deletions(-)

diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index 796098c7740..76316d97059 100644
--- a/docs/source/user-guide/latest/datasources.md
+++ b/docs/source/user-guide/latest/datasources.md
@@ -481,7 +481,7 @@ hadoop-azure must be on the driver classpath. It already is whenever Spark can l
 
 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.
 
 ### Environment variables
 
diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs
index 2dbf6712150..f686f05d619 100644
--- a/native/core/src/parquet/objectstore/azure.rs
+++ b/native/core/src/parquet/objectstore/azure.rs
@@ -37,8 +37,9 @@
 //! - `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 the `declined` and `error` arms, or the providers `resolved_settings`
-//! rejects: the driver plans those on Spark instead. Native writes still fail with them.
+//! Scans never reach the `declined` and `error` arms, the providers `resolved_settings` rejects,
+//! or a client credentials endpoint that is not `https` or has no tenant segment: the driver plans
+//! those on Spark instead. Native writes still fail with them.
 //!
 //! `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
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 ef9601c4b3e..5526388d167 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
@@ -472,9 +472,10 @@ private[comet] object AbfsAuthResolver extends Logging {
   /**
    * 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 built-in OAuth provider with no
-   * `object_store` counterpart, or a resolution that failed. 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.
+   * `object_store` counterpart, a client credentials endpoint the native store rejects, or a
+   * resolution that failed. 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.
    */
   def nativeFallbackReason(uri: URI, outcome: Outcome): Option[String] = {
     // The container and host only: a password in the user info stays out of the reason.
@@ -496,6 +497,13 @@ private[comet] object AbfsAuthResolver extends Logging {
             Some(c @ (Providers.REFRESH_TOKEN | Providers.USER_PASSWORD)),
             _) =>
         Some(unsupported(s"auth type ${AuthTypes.OAUTH} with provider $c"))
+      case Resolved(AuthTypes.OAUTH, Some(c @ Providers.CLIENT_CREDS), values)
+          if !values.get(Keys.CLIENT_ENDPOINT).exists(clientCredsEndpointUsable) =>
+        // The endpoint stays out of the reason: it may carry user info.
+        Some(
+          unsupported(
+            s"auth type ${AuthTypes.OAUTH} with provider $c and an endpoint that is not an " +
+              "https URL with a tenant path segment"))
       case Failed(_, message) =>
         // Some of Hadoop's own messages quote the whole URI, user info included.
         val safeMessage =
@@ -505,6 +513,23 @@ private[comet] object AbfsAuthResolver extends Logging {
     }
   }
 
+  /**
+   * Whether the native store can post client credentials token requests to `endpoint`: an `https`
+   * URL with a host and a tenant segment, which `client_creds_settings` in `azure.rs` reads the
+   * same way (the tenant is the segment before the last `oauth2`, else the first). Conservative
+   * where the native URL parser and `java.net.URI` could disagree: a percent escape, a backslash
+   * or a dot segment is declined, since Hadoop's own reader may take it.
+   */
+  private[objectstore] def clientCredsEndpointUsable(endpoint: String): Boolean =
+    Try(new URI(endpoint)).toOption.exists { uri =>
+      val path = Option(uri.getRawPath).getOrElse("")
+      val segments = path.stripPrefix("/").split("/", -1)
+      val tenantAt = Some(segments.lastIndexOf("oauth2")).filter(_ > 0).map(_ - 1).getOrElse(0)
+      "https".equalsIgnoreCase(uri.getScheme) && uri.getHost != null &&
+      !path.exists(c => c == '%' || c == '\\') && !segments.exists(s => s == "." || s == "..") &&
+      segments(tenantAt).nonEmpty
+    }
+
   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 fbdffa35a42..b6f63ac77ef 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala
@@ -342,7 +342,7 @@ object NativeConfig {
 
     import scala.jdk.CollectionConverters._
     val options = scala.collection.mutable.Map[String, String]() ++= schemeListOptions(hadoopConf)
-    val s3CompliantSchemes = parseSchemeSet(hadoopConf.get(COMET_S3_COMPLIANT_SCHEMES_KEY))
+    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).
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 0c4aed1b931..2c53fc71df6 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
@@ -850,6 +850,68 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
       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 native rejects falls back") {
+    // `azure.rs` builds no store for an `http` endpoint or one without a tenant segment.
+    Seq(
+      "http://login.microsoftonline.com/tenant/oauth2/token",
+      "https://login.microsoftonline.com/",
+      "https://login.microsoftonline.com").foreach { endpoint =>
+      withClue(s"$endpoint: ") {
+        assertFallsBack(clientCredsConf(endpoint), Providers.CLIENT_CREDS, "tenant path segment")
+      }
+    }
+  }
+
+  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("clientCredsEndpointUsable - agrees with the endpoints azure.rs builds a store for") {
+    // The cases of `oauth_endpoint_parts_split_tenant_and_authority_host` in azure.rs, over https.
+    val usable = Seq(
+      "https://login.microsoftonline.com/tenant/oauth2/token",
+      "https://user:pass@auth-proxy.example/t/oauth2/token",
+      "https://auth-proxy.example:8443/t/oauth2/token",
+      "https://[::1]:8443/t/oauth2/token",
+      "https://auth-proxy.example/aad/t/oauth2/v2.0/token",
+      "https://auth-proxy.example/aad/t/token",
+      "https://auth-proxy.example/oauth2/token",
+      "https://h//t/oauth2/token",
+      "https://h/aad/t/oauth2/v2.0/token?q=oauth2#f",
+      "https://h/a/oauth2/b/oauth2/token",
+      "HTTPS://h/t/oauth2/token")
+    val unusable = Seq(
+      "http://login.microsoftonline.com/tenant/oauth2/token",
+      "https://auth-proxy.example/",
+      "https://auth-proxy.example",
+      "https://h/a//oauth2/token",
+      "file:///t/oauth2/token",
+      "urn:t:oauth2",
+      "not a url",
+      // Resolved differently by the native URL parser, so declined rather than guessed at.
+      "https://h/a/%2e%2e/t/oauth2/token",
+      "https://h/a\\b/t/oauth2/token",
+      "https://h/a/../oauth2/token")
+    usable.foreach(e => withClue(e)(AbfsAuthResolver.clientCredsEndpointUsable(e) shouldBe true))
+    unusable.foreach(e =>
+      withClue(e)(AbfsAuthResolver.clientCredsEndpointUsable(e) shouldBe false))
+  }
+
   test("extractScanObjectStoreOptions - a failed ABFS resolution falls back to Spark") {
     assertFallsBack(
       abfsScanConf(Keys.AUTH_TYPE -> "NotAnAuthType"),

From 5ee970e6c528544e2a528b01ea7f4126abd2e1ac Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Mon, 26 Oct 2026 09:36:24 -0600
Subject: [PATCH 5/6] fix: fall back for an http Workload Identity authority in
 the native ABFS scan

object_store's token client refuses http, so a Resolved WorkloadIdentityTokenProvider
whose fs.azure.account.oauth2.msi.authority is not https plans natively and fails
every task, while Hadoop posts to it. nativeFallbackReason now declines it. The
reason leaves the authority out because it can carry user info.
---
 docs/source/user-guide/latest/datasources.md  |  2 +-
 native/core/src/parquet/objectstore/azure.rs  |  5 ++-
 .../comet/objectstore/AbfsAuthResolver.scala  | 19 +++++++++-
 .../comet/objectstore/NativeConfigSuite.scala | 38 +++++++++++++++++++
 4 files changed, 60 insertions(+), 4 deletions(-)

diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index 76316d97059..69695939515 100644
--- a/docs/source/user-guide/latest/datasources.md
+++ b/docs/source/user-guide/latest/datasources.md
@@ -481,7 +481,7 @@ hadoop-azure must be on the driver classpath. It already is whenever Spark can l
 
 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, or one with no tenant segment, makes the scan fall back to Spark's own scan.
+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`).
 
 ### Environment variables
 
diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs
index f686f05d619..d25d8fc3da6 100644
--- a/native/core/src/parquet/objectstore/azure.rs
+++ b/native/core/src/parquet/objectstore/azure.rs
@@ -38,8 +38,9 @@
 //! - missing or anything else: an error, so a plan that did not run the resolver cannot pass.
 //!
 //! Scans never reach the `declined` and `error` arms, the providers `resolved_settings` rejects,
-//! or a client credentials endpoint that is not `https` or has no tenant segment: the driver plans
-//! those on Spark instead. Native writes still fail with them.
+//! a client credentials endpoint that is not `https` or has no tenant segment, or a Workload
+//! Identity authority that is not `https`: the driver plans those on Spark instead. Native writes
+//! still fail with them.
 //!
 //! `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
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 5526388d167..ce32d1f83b5 100644
--- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
+++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala
@@ -504,6 +504,12 @@ private[comet] object AbfsAuthResolver extends Logging {
           unsupported(
             s"auth type ${AuthTypes.OAUTH} with provider $c and an endpoint that is not an " +
               "https URL with a tenant path segment"))
+      case Resolved(AuthTypes.OAUTH, Some(c @ Providers.WORKLOAD_IDENTITY), values)
+          if !values.get(Keys.MSI_AUTHORITY).exists(httpsHost) =>
+        Some(
+          unsupported(
+            s"auth type ${AuthTypes.OAUTH} with provider $c and an authority that is not an " +
+              "https URL"))
       case Failed(_, message) =>
         // Some of Hadoop's own messages quote the whole URI, user info included.
         val safeMessage =
@@ -525,11 +531,22 @@ private[comet] object AbfsAuthResolver extends Logging {
       val path = Option(uri.getRawPath).getOrElse("")
       val segments = path.stripPrefix("/").split("/", -1)
       val tenantAt = Some(segments.lastIndexOf("oauth2")).filter(_ > 0).map(_ - 1).getOrElse(0)
-      "https".equalsIgnoreCase(uri.getScheme) && uri.getHost != null &&
+      hasHttpsHost(uri) &&
       !path.exists(c => c == '%' || c == '\\') && !segments.exists(s => s == "." || s == "..") &&
       segments(tenantAt).nonEmpty
     }
 
+  /**
+   * Whether the native store can post Workload Identity token requests to `authority`. It adds
+   * `/<tenant>/oauth2/v2.0/token` to the authority, as Hadoop does, over a client that refuses
+   * `http`, so the authority must be an `https` URL with a host.
+   */
+  private[objectstore] def httpsHost(authority: String): Boolean =
+    Try(new URI(authority)).toOption.exists(hasHttpsHost)
+
+  private def hasHttpsHost(uri: URI): Boolean =
+    "https".equalsIgnoreCase(uri.getScheme) && uri.getHost != null
+
   private def providerMarker(authType: String): String =
     if (authType == AuthTypes.SAS) Markers.SAS_PROVIDER_CLASS else Markers.OAUTH_PROVIDER_CLASS
 
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 2c53fc71df6..641643679ab 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
@@ -881,6 +881,44 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
     }
   }
 
+  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/"),
+      Providers.WORKLOAD_IDENTITY,
+      "authority that 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("httpsHost - needs an https URL with a host") {
+    Seq("https://login.microsoftonline.com/", "https://h:8443/aad", "HTTPS://h").foreach(a =>
+      withClue(a)(AbfsAuthResolver.httpsHost(a) shouldBe true))
+    Seq(
+      "http://login.microsoftonline.com/",
+      "https:///",
+      "login.microsoftonline.com",
+      "not a url")
+      .foreach(a => withClue(a)(AbfsAuthResolver.httpsHost(a) shouldBe false))
+  }
+
   test("clientCredsEndpointUsable - agrees with the endpoints azure.rs builds a store for") {
     // The cases of `oauth_endpoint_parts_split_tenant_and_authority_host` in azure.rs, over https.
     val usable = Seq(

From beda9508f907a6b00bf94bb30c265b7b95295e65 Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Tue, 27 Oct 2026 05:27:47 -0600
Subject: [PATCH 6/6] fix: ask native whether it can build the ABFS store
 instead of copying its rules

The driver now calls NativeBase.abfsStoreError, which runs azure::create_store on the
resolved options, and falls back with native's own message when it fails. The Scala
copies of the client credentials endpoint and Workload Identity authority rules are
removed. Native rejects an http endpoint or authority when it builds the store, so
native writes get that error up front too.
---
 docs/source/user-guide/latest/datasources.md  |  2 +-
 native/core/src/lib.rs                        | 32 ++++++-
 native/core/src/parquet/objectstore/azure.rs  | 92 +++++++++++++++++--
 .../java/org/apache/comet/NativeBase.java     | 13 +++
 .../comet/objectstore/AbfsAuthResolver.scala  | 72 +++++----------
 .../comet/objectstore/NativeConfigSuite.scala | 73 +++++----------
 6 files changed, 173 insertions(+), 111 deletions(-)

diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md
index 69695939515..a24cc12a8ce 100644
--- a/docs/source/user-guide/latest/datasources.md
+++ b/docs/source/user-guide/latest/datasources.md
@@ -481,7 +481,7 @@ hadoop-azure must be on the driver classpath. It already is whenever Spark can l
 
 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, 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`).
+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
 
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<Config> {
     let console_append = ConsoleAppender::builder()
diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs
index d25d8fc3da6..8b18ac70250 100644
--- a/native/core/src/parquet/objectstore/azure.rs
+++ b/native/core/src/parquet/objectstore/azure.rs
@@ -37,10 +37,10 @@
 //! - `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 the `declined` and `error` arms, the providers `resolved_settings` rejects,
-//! a client credentials endpoint that is not `https` or has no tenant segment, or a Workload
-//! Identity authority that is not `https`: the driver plans those on Spark instead. Native writes
-//! still fail with them.
+//! 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
@@ -67,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
@@ -134,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,
@@ -237,6 +254,7 @@ fn client_creds_settings(
              before `oauth2`"
         ))
     })?;
+    require_https(HADOOP_OAUTH_CLIENT_ENDPOINT, &host)?;
     Ok(vec![
         (
             AzureConfigKey::ClientId,
@@ -279,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,
@@ -292,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!(
@@ -534,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(
@@ -568,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(),
@@ -699,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 ce32d1f83b5..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
 
 /**
@@ -471,11 +472,11 @@ private[comet] object AbfsAuthResolver extends Logging {
 
   /**
    * 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 built-in OAuth provider with no
-   * `object_store` counterpart, a client credentials endpoint the native store rejects, or a
-   * resolution that failed. All are known on the driver, so the scan falls back to Spark's own
+   * 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.
+   * 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.
@@ -485,67 +486,40 @@ private[comet] object AbfsAuthResolver extends Logging {
       if (at < 0) authority
       else authority.substring(0, at).takeWhile(_ != ':') + authority.substring(at)
     val location = s"${uri.getScheme}://$safeAuthority"
-    def unsupported(mechanism: String): String =
-      s"Native scan does not support ABFS authentication $mechanism for $location"
     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(unsupported(s"auth type $authType$classes"))
-      case Resolved(
-            AuthTypes.OAUTH,
-            Some(c @ (Providers.REFRESH_TOKEN | Providers.USER_PASSWORD)),
-            _) =>
-        Some(unsupported(s"auth type ${AuthTypes.OAUTH} with provider $c"))
-      case Resolved(AuthTypes.OAUTH, Some(c @ Providers.CLIENT_CREDS), values)
-          if !values.get(Keys.CLIENT_ENDPOINT).exists(clientCredsEndpointUsable) =>
-        // The endpoint stays out of the reason: it may carry user info.
         Some(
-          unsupported(
-            s"auth type ${AuthTypes.OAUTH} with provider $c and an endpoint that is not an " +
-              "https URL with a tenant path segment"))
-      case Resolved(AuthTypes.OAUTH, Some(c @ Providers.WORKLOAD_IDENTITY), values)
-          if !values.get(Keys.MSI_AUTHORITY).exists(httpsHost) =>
-        Some(
-          unsupported(
-            s"auth type ${AuthTypes.OAUTH} with provider $c and an authority that is not an " +
-              "https URL"))
+          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 _: Resolved | Unconfigured => None
+      case Unconfigured => None
     }
   }
 
   /**
-   * Whether the native store can post client credentials token requests to `endpoint`: an `https`
-   * URL with a host and a tenant segment, which `client_creds_settings` in `azure.rs` reads the
-   * same way (the tenant is the segment before the last `oauth2`, else the first). Conservative
-   * where the native URL parser and `java.net.URI` could disagree: a percent escape, a backslash
-   * or a dot segment is declined, since Hadoop's own reader may take it.
+   * 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[objectstore] def clientCredsEndpointUsable(endpoint: String): Boolean =
-    Try(new URI(endpoint)).toOption.exists { uri =>
-      val path = Option(uri.getRawPath).getOrElse("")
-      val segments = path.stripPrefix("/").split("/", -1)
-      val tenantAt = Some(segments.lastIndexOf("oauth2")).filter(_ > 0).map(_ - 1).getOrElse(0)
-      hasHttpsHost(uri) &&
-      !path.exists(c => c == '%' || c == '\\') && !segments.exists(s => s == "." || s == "..") &&
-      segments(tenantAt).nonEmpty
+  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
     }
-
-  /**
-   * Whether the native store can post Workload Identity token requests to `authority`. It adds
-   * `/<tenant>/oauth2/v2.0/token` to the authority, as Hadoop does, over a client that refuses
-   * `http`, so the authority must be an `https` URL with a host.
-   */
-  private[objectstore] def httpsHost(authority: String): Boolean =
-    Try(new URI(authority)).toOption.exists(hasHttpsHost)
-
-  private def hasHttpsHost(uri: URI): Boolean =
-    "https".equalsIgnoreCase(uri.getScheme) && uri.getHost != null
+  }
 
   private def providerMarker(authType: String): String =
     if (authType == AuthTypes.SAS) Markers.SAS_PROVIDER_CLASS else Markers.OAUTH_PROVIDER_CLASS
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 641643679ab..8c4ad9256ef 100644
--- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala
@@ -858,15 +858,28 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
       Keys.CLIENT_ID -> "client-123",
       Keys.CLIENT_SECRET -> "client-secret-value")
 
-  test(
-    "extractScanObjectStoreOptions - a client credentials endpoint native rejects falls back") {
-    // `azure.rs` builds no store for an `http` endpoint or one without a tenant segment.
+  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(
-      "http://login.microsoftonline.com/tenant/oauth2/token",
-      "https://login.microsoftonline.com/",
-      "https://login.microsoftonline.com").foreach { endpoint =>
+      "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: ") {
-        assertFallsBack(clientCredsConf(endpoint), Providers.CLIENT_CREDS, "tenant path segment")
+        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")
+        }
       }
     }
   }
@@ -893,8 +906,8 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
         Keys.MSI_AUTHORITY -> authority)
     assertFallsBack(
       conf("http://user:authority-password@login.example/"),
-      Providers.WORKLOAD_IDENTITY,
-      "authority that is not an https URL")
+      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/"),
@@ -908,48 +921,6 @@ class NativeConfigSuite extends AnyFunSuite with Matchers {
     }
   }
 
-  test("httpsHost - needs an https URL with a host") {
-    Seq("https://login.microsoftonline.com/", "https://h:8443/aad", "HTTPS://h").foreach(a =>
-      withClue(a)(AbfsAuthResolver.httpsHost(a) shouldBe true))
-    Seq(
-      "http://login.microsoftonline.com/",
-      "https:///",
-      "login.microsoftonline.com",
-      "not a url")
-      .foreach(a => withClue(a)(AbfsAuthResolver.httpsHost(a) shouldBe false))
-  }
-
-  test("clientCredsEndpointUsable - agrees with the endpoints azure.rs builds a store for") {
-    // The cases of `oauth_endpoint_parts_split_tenant_and_authority_host` in azure.rs, over https.
-    val usable = Seq(
-      "https://login.microsoftonline.com/tenant/oauth2/token",
-      "https://user:pass@auth-proxy.example/t/oauth2/token",
-      "https://auth-proxy.example:8443/t/oauth2/token",
-      "https://[::1]:8443/t/oauth2/token",
-      "https://auth-proxy.example/aad/t/oauth2/v2.0/token",
-      "https://auth-proxy.example/aad/t/token",
-      "https://auth-proxy.example/oauth2/token",
-      "https://h//t/oauth2/token",
-      "https://h/aad/t/oauth2/v2.0/token?q=oauth2#f",
-      "https://h/a/oauth2/b/oauth2/token",
-      "HTTPS://h/t/oauth2/token")
-    val unusable = Seq(
-      "http://login.microsoftonline.com/tenant/oauth2/token",
-      "https://auth-proxy.example/",
-      "https://auth-proxy.example",
-      "https://h/a//oauth2/token",
-      "file:///t/oauth2/token",
-      "urn:t:oauth2",
-      "not a url",
-      // Resolved differently by the native URL parser, so declined rather than guessed at.
-      "https://h/a/%2e%2e/t/oauth2/token",
-      "https://h/a\\b/t/oauth2/token",
-      "https://h/a/../oauth2/token")
-    usable.foreach(e => withClue(e)(AbfsAuthResolver.clientCredsEndpointUsable(e) shouldBe true))
-    unusable.foreach(e =>
-      withClue(e)(AbfsAuthResolver.clientCredsEndpointUsable(e) shouldBe false))
-  }
-
   test("extractScanObjectStoreOptions - a failed ABFS resolution falls back to Spark") {
     assertFallsBack(
       abfsScanConf(Keys.AUTH_TYPE -> "NotAnAuthType"),