pan3793 commented on code in PR #59697:
URL: https://github.com/apache/spark/pull/59697#discussion_r4346497268
##########
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:
Fixed in 2f13da322f4: the metadata info now falls back to the
class-name-only info on `LinkageError` too, including from the loadability
check, and doesn't touch the class of a function with `USING` resources before
the first invocation loads them. The invocation path (`loadFunctionClass`) is
unchanged from before this PR and is shared with Hive UDFs, so I'm leaving it
as is here.
##########
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])) {
+ try {
+ val clazz = Utils.classForName(func.className)
+ makeExprInfoForV2Function(func, newV2FunctionInstance(clazz,
func.identifier))
+ } catch {
+ case NonFatal(_) => makeExprInfoForHiveFunction(func)
Review Comment:
Fixed in 2f13da322f4, along with the comment above. Added tests for a class
with a failing static initializer and for `USING` resources loading before the
class is initialized.
##########
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:
Fixed in 2f13da322f4: the builder tags the expression with `FUNC_ALIAS`,
like Hive UDF wrappers do. Added tests for DISTINCT / FILTER / IGNORE NULLS /
WITHIN GROUP errors and column names. The magic-method shape keeps its
`invoke(...)` column name, same as the `FunctionCatalog` path.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala:
##########
@@ -1442,6 +1442,17 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
new FunctionAlreadyExistsException(func.nameParts)
}
+ def cannotInstantiateFunctionClassError(
+ className: String, func: FunctionIdentifier, cause: Throwable):
Throwable = {
+ new AnalysisException(
+ errorClass = "CANNOT_INSTANTIATE_FUNCTION_CLASS",
+ messageParameters = Map(
+ "className" -> className,
+ "functionName" -> toSQLId(func.toString),
+ "msg" -> Option(cause.getMessage).getOrElse("")),
Review Comment:
Fixed in 2f13da322f4: the constructor's exception is unwrapped from
`InvocationTargetException` and reported as `<msg>`, and the message now covers
a throwing constructor. Tests assert all parameters for the temporary and
persistent paths.
##########
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:
This is a pre-existing issue: V2 functions served by a `FunctionCatalog`
from a jar added at runtime hit the same codegen class resolution. This feature
makes it easier to trigger, but the fix belongs in how generated code resolves
classes, so I'd keep it out of this PR. Documented it as a known limitation in
the PR description.
--
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]
