LuciferYang commented on code in PR #59697:
URL: https://github.com/apache/spark/pull/59697#discussion_r4345918710
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2076,14 +2079,47 @@ class SessionCatalog(
* Constructs a [[FunctionBuilder]] based on the provided function metadata.
*/
private def makeFunctionBuilder(func: CatalogFunction): FunctionBuilder = {
+ val clazz = loadFunctionClass(func)
+ if (isV2FunctionClass(clazz)) {
+ // A DataSource V2 function: bind it to the arguments at each call site.
An aggregate
+ // function is returned as a bare `V2Aggregator`, which the caller wraps
into an
+ // `AggregateExpression` carrying DISTINCT / FILTER. This check goes
first so that a class
+ // also implementing a Hive UDF interface, for other engines sharing the
metastore, runs as
+ // a V2 function in Spark.
+ val unbound = newV2FunctionInstance(clazz, func.identifier)
+ (input: Seq[Expression]) => V2ExpressionUtils.bindFunction(unbound,
input)
Review Comment:
For a V2 function loaded from a jar that implements the magic method,
`resolveScalarFunction` produces `Invoke` / `StaticInvoke`, and the generated
code names the user class. Janino resolves that name through the current
thread's context classloader, but `ADD JAR` and `USING JAR` only switch the
context classloader of the thread that loads the jar to
`sharedState.jarClassLoader`. An existing pool thread of a multi-threaded
driver, or the next line in spark-shell (the REPL restores the loader after
each line), can't see the class at compile time. With default configs,
whole-stage codegen only logs a WARN for the compile failure and that stage
runs operator by operator. `TakeOrderedAndProjectExec` builds
`LazilyGeneratedOrdering` on the driver with no fallback, so `SELECT c FROM t
ORDER BY f(c) LIMIT 10` fails.
The "persistent function USING JAR" test avoids this with `magic = false`.
I'd like to hear your take on the fix. One option is to make the execution-time
context classloader see `jarClassLoader`, e.g. by reusing
`artifactManager.withResources`. In classic mode, though, it only switches the
loader when `alwaysApplyClassloader` is on or the session has added artifacts
through `addArtifact` (`ADD JAR` doesn't count), and subquery/broadcast threads
don't go through it. Whichever way you go, a test with `magic = true` that
restores the original context classloader before calling the function would
catch this.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2167,12 +2203,49 @@ class SessionCatalog(
if (funcDefinition.isUserDefinedFunction) {
UserDefinedFunction.fromCatalogFunction(funcDefinition,
parser).toExpressionInfo
} else {
- makeExprInfoForHiveFunction(funcDefinition)
+ makeExprInfoForClassFunction(funcDefinition)
}
}
registry.registerFunction(identToRegister, resolvedInfo, functionBuilder)
}
+ /**
+ * Builds the [[ExpressionInfo]] of a class-based function (`CREATE FUNCTION
... AS 'class'`).
+ * A DataSource V2 function reports its own description, which requires
instantiating it; if the
+ * class is not on the classpath yet (its `USING` resources are only loaded
on first invocation)
+ * or cannot be instantiated, fall back to the class-name-only info, so that
like a Hive UDF, a
+ * broken function only fails when invoked. Other classes are checked
without being initialized.
+ */
+ private def makeExprInfoForClassFunction(func: CatalogFunction):
ExpressionInfo = {
+ if (Utils.classIsLoadableAndAssignableFrom(func.className,
classOf[UnboundFunction])) {
Review Comment:
`makeExprInfoForClassFunction` now loads the class of every persistent class
function, Hive UDFs included, but neither the `Try` inside
`classIsLoadableAndAssignableFrom` nor the `case NonFatal(_)` here catches
`LinkageError`. If a class is found but fails to link or initialize (missing
superclass, compiled for a newer JDK, a static initializer that throws), errors
like `NoClassDefFoundError` escape. `listFunctions()` goes through this for
every user function, so one such function fails the whole listing, and
`DESCRIBE FUNCTION` / `functionExists` throw a raw Error. Before this change
the metadata path loaded no classes and all of these worked, as the PR
description says they should ("not on ... DESCRIBE or listing").
Could you wrap the whole body, including the
`classIsLoadableAndAssignableFrom` call, so that a `LinkageError` falls back to
`makeExprInfoForHiveFunction` the same way `NonFatal` does? After that, a
failing static initializer throws from `Utils.classForName` in
`loadFunctionClass` before `newV2FunctionInstance` runs, so reporting it as an
`AnalysisException` rather than a raw Error means wrapping that call too. Hive
UDFs share it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
