pan3793 commented on code in PR #59697:
URL: https://github.com/apache/spark/pull/59697#discussion_r4374766527
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2073,17 +2077,63 @@ class SessionCatalog(
}
/**
- * Constructs a [[FunctionBuilder]] based on the provided function metadata.
+ * Constructs a [[FunctionBuilder]] and the [[ExpressionInfo]] based on the
provided function
+ * metadata.
*/
- private def makeFunctionBuilder(func: CatalogFunction): FunctionBuilder = {
+ private def makeFunctionBuilderAndInfo(
+ func: CatalogFunction): (FunctionBuilder, ExpressionInfo) = {
+ val clazz = loadFunctionClass(func)
+ val name = func.identifier.unquotedString
+ 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. The alias makes errors and column names use
the function name,
+ // like the Hive UDF wrappers do.
+ val unbound = newV2FunctionInstance(clazz, func.identifier)
+ val builder = (input: Seq[Expression]) => {
+ val expr = V2ExpressionUtils.bindFunction(unbound, input)
+ expr.setTagValue(FunctionRegistry.FUNC_ALIAS, name)
+ expr
+ }
+ (builder, makeExprInfoForV2Function(func, unbound))
Review Comment:
@dongjoon-hyun Fixed in f1af9b84046: a throwing `description()` falls back
to no description on every path, including registration. Added a test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2073,17 +2077,63 @@ class SessionCatalog(
}
/**
- * Constructs a [[FunctionBuilder]] based on the provided function metadata.
+ * Constructs a [[FunctionBuilder]] and the [[ExpressionInfo]] based on the
provided function
+ * metadata.
*/
- private def makeFunctionBuilder(func: CatalogFunction): FunctionBuilder = {
+ private def makeFunctionBuilderAndInfo(
+ func: CatalogFunction): (FunctionBuilder, ExpressionInfo) = {
+ val clazz = loadFunctionClass(func)
+ val name = func.identifier.unquotedString
+ 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. The alias makes errors and column names use
the function name,
+ // like the Hive UDF wrappers do.
+ val unbound = newV2FunctionInstance(clazz, func.identifier)
+ val builder = (input: Seq[Expression]) => {
+ val expr = V2ExpressionUtils.bindFunction(unbound, input)
+ expr.setTagValue(FunctionRegistry.FUNC_ALIAS, name)
+ expr
+ }
+ (builder, makeExprInfoForV2Function(func, unbound))
+ } else {
+ val builder = (input: Seq[Expression]) =>
+ functionExpressionBuilder.makeExpression(name, clazz, input)
+ (builder, makeExprInfoForHiveFunction(func))
+ }
+ }
+
+ /**
+ * Loads the class a class-based function (`CREATE FUNCTION ... AS 'class'`)
refers to.
+ */
+ private def loadFunctionClass(func: CatalogFunction): Class[_] = {
val className = func.className
if (!Utils.classIsLoadable(className)) {
throw QueryCompilationErrors.cannotLoadClassWhenRegisteringFunctionError(
className, func.identifier)
}
- val clazz = Utils.classForName(className)
- val name = func.identifier.unquotedString
- (input: Seq[Expression]) => functionExpressionBuilder.makeExpression(name,
clazz, input)
+ Utils.classForName(className)
+ }
+
+ private def isV2FunctionClass(clazz: Class[_]): Boolean = {
+ classOf[UnboundFunction].isAssignableFrom(clazz)
+ }
+
+ private def newV2FunctionInstance(
+ clazz: Class[_],
+ ident: FunctionIdentifier): UnboundFunction = {
+ try {
+
clazz.getDeclaredConstructor().newInstance().asInstanceOf[UnboundFunction]
+ } catch {
+ case e: InvocationTargetException =>
Review Comment:
@dongjoon-hyun Fixed in f1af9b84046: only a non-fatal or linkage cause is
wrapped; other causes are rethrown.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2073,17 +2077,63 @@ class SessionCatalog(
}
/**
- * Constructs a [[FunctionBuilder]] based on the provided function metadata.
+ * Constructs a [[FunctionBuilder]] and the [[ExpressionInfo]] based on the
provided function
+ * metadata.
*/
- private def makeFunctionBuilder(func: CatalogFunction): FunctionBuilder = {
+ private def makeFunctionBuilderAndInfo(
+ func: CatalogFunction): (FunctionBuilder, ExpressionInfo) = {
+ val clazz = loadFunctionClass(func)
+ val name = func.identifier.unquotedString
+ 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. The alias makes errors and column names use
the function name,
+ // like the Hive UDF wrappers do.
+ val unbound = newV2FunctionInstance(clazz, func.identifier)
+ val builder = (input: Seq[Expression]) => {
+ val expr = V2ExpressionUtils.bindFunction(unbound, input)
Review Comment:
@dongjoon-hyun Fixed in f1af9b84046: `bindFunction` takes the registered
identifier for its errors on this path; the `FunctionCatalog` path is unchanged.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2720,7 +2816,7 @@ class SessionCatalog(
val info = if (funcMetadata.isUserDefinedFunction) {
UserDefinedFunction.fromCatalogFunction(funcMetadata,
parser).toExpressionInfo
} else {
- makeExprInfoForHiveFunction(funcMetadata)
+ makeExprInfoForClassFunction(funcMetadata)
Review Comment:
@dongjoon-hyun Fixed in 45f1d6bdf84: `V1Function` builds its info lazily, so
resolving and invoking a persistent function constructs the class once.
--
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]
