andygrove commented on code in PR #7221:
URL: https://github.com/apache/datafusion-comet/pull/7221#discussion_r4392164824
##########
native/core/src/lib.rs:
##########
@@ -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);
Review Comment:
Done. Both arrays are `JObjectArray<JString>` now, so `get_element` returns
a `JString` and the two `unsafe` casts are gone. Pushed in 240e47ed48.
##########
spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala:
##########
@@ -467,6 +470,57 @@ private[comet] object AbfsAuthResolver extends Logging {
Markers.ERROR_MESSAGE -> message)
}
+ /**
+ * Why the native store cannot read `uri` under `outcome`, or None when it
can: a mechanism
+ * Hadoop named that runs Java code per request, a resolution that failed,
or resolved values
+ * the native store rejects. All are known on the driver, so the scan falls
back to Spark's own
+ * reader, which authenticates through hadoop-azure, instead of planning
natively and failing
+ * every task. For `Resolved`, native itself answers (see
[[nativeStoreError]]), so the rules
+ * live in one place.
+ */
+ def nativeFallbackReason(uri: URI, outcome: Outcome): Option[String] = {
+ // The container and host only: a password in the user info stays out of
the reason.
+ val authority = Option(uri.getRawAuthority).getOrElse("")
+ val at = authority.lastIndexOf('@')
+ val safeAuthority =
+ if (at < 0) authority
+ else authority.substring(0, at).takeWhile(_ != ':') +
authority.substring(at)
+ val location = s"${uri.getScheme}://$safeAuthority"
+ outcome match {
+ case Declined(authType, providerClass, assertionProviderClass) =>
+ val classes = providerClass.map(c => s" with provider
$c").getOrElse("") +
+ assertionProviderClass.map(c => s" and client assertion provider
$c").getOrElse("")
+ Some(
+ s"Native scan does not support ABFS authentication auth type
$authType$classes " +
+ s"for $location")
+ case resolved: Resolved =>
+ // The message names option keys, never their values.
+ nativeStoreError(s"$location/", toOptions(resolved)).map(message =>
+ s"Native scan cannot use the ABFS authentication for $location:
$message")
+ case Failed(_, message) =>
+ // Some of Hadoop's own messages quote the whole URI, user info
included.
+ val safeMessage =
+ if (authority.isEmpty) message else message.replace(authority,
safeAuthority)
+ Some(s"Hadoop ABFS authentication failed on the driver for $location:
$safeMessage")
+ case Unconfigured => None
+ }
+ }
+
+ /**
+ * Why native cannot build the store for `url` from `options`, from the same
code the scan runs.
+ * When native cannot be asked (the library is not loaded), this assumes it
can: such a build
+ * cannot run the native scan anyway.
+ */
+ private def nativeStoreError(url: String, options: Map[String, String]):
Option[String] = {
+ val (keys, values) = options.toArray.unzip
+ try Option(NativeBase.abfsStoreError(url, keys, values))
+ catch {
+ case e @ (NonFatal(_) | _: LinkageError) =>
+ logDebug(s"Could not ask native whether it can build the ABFS store
for $url", e)
+ None
+ }
Review Comment:
You are right: a panic in `create_store` would have planned natively and
failed every task, with only a `logDebug` on the driver. Only a `LinkageError`
(library not loaded) assumes the store builds now. Any other failure becomes a
fallback reason, "could not ask native whether it can build the ABFS store:
<class>: <message>", logged at warn and shown in the explain output. I made the
JNI call a parameter of `nativeStoreError` so tests can make it fail.
`NativeConfigSuite` has a test for each case: native's answer passed through
(and null meaning the store builds), an `UnsatisfiedLinkError` assuming it
builds, and a `RuntimeException` giving the fallback reason.
`NativeConfigSuite`, `AbfsAuthResolverSuite`, `AbfsAuthParitySuite` and
`AbfsScanFallbackSuite` pass on the default Spark 4.1 profile with the native
library loaded. Pushed in 240e47ed48.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
