LuciferYang commented on code in PR #59697:
URL: https://github.com/apache/spark/pull/59697#discussion_r4441625156
##########
docs/sql-ref-syntax-ddl-create-function.md:
##########
@@ -80,13 +80,22 @@ CREATE [ OR REPLACE ] [ TEMPORARY ] FUNCTION [ IF NOT
EXISTS ]
* **class_name**
Specifies the name of the class that provides the implementation for
function to be created.
- The implementing class should extend one of the base classes as follows:
+ The implementing class should have a public no-arg constructor, and extend
one of the base
+ classes as follows:
+ * Should implement `UnboundFunction` in
`org.apache.spark.sql.connector.catalog.functions`
+ package.
* Should extend `UDF` or `UDAF` in `org.apache.hadoop.hive.ql.exec`
package.
* Should extend `AbstractGenericUDAFResolver`, `GenericUDF`, or
`GenericUDTF` in `org.apache.hadoop.hive.ql.udf.generic` package.
* Should extend `UserDefinedAggregateFunction` in
`org.apache.spark.sql.expressions` package.
+ **Note:** If an `UnboundFunction` binds to a `ScalarFunction` that
implements the magic method
+ `invoke`, and its jar is added at runtime (`ADD JAR`, or `USING JAR`),
queries calling it may
+ fail to compile generated code when run on a thread other than the one
that added the jar,
Review Comment:
The trigger here, "on a thread other than the one that added the jar",
doesn't match its own example: a later line in spark-shell runs on the same
thread, but the REPL resets the context classloader after each line. What
matters is whether the context classloader of the driver thread compiling the
query can see the jar. The note also says only "may fail to compile generated
code". With default configs, whole-stage codegen logs the compile error and
falls back, so results stay correct. Queries that hit an operator without a
fallback, such as `ORDER BY f(c) LIMIT 10`, fail. The PR description covers
both cases, and it would help to have them here too.
Could you open a JIRA for how generated code resolves user classes and link
it from this note? A test with `magic = true` that restores the context
classloader and then calls the function would also pin down the current
behavior. In tests the whole-stage fallback is disabled, so even a plain
`SELECT f(c)` fails there.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2073,17 +2077,69 @@ 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 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, displayName = Some(func.identifier))
+ expr.setTagValue(FunctionRegistry.FUNC_ALIAS, name)
Review Comment:
A magic method that is an instance method binds to `Invoke(Literal(<function
instance>))`, and `Invoke.sql` doesn't use `prettyName`, so the FUNC_ALIAS set
here doesn't reach the column name. The name is `invoke(<class>@<identity
hash>.invoke(s))`, and when `bind` returns the function itself, as
`SimpleFunction` does, the hash changes with each registration. I tried this:
after `CREATE TEMPORARY VIEW v AS SELECT f(s) FROM t`, running `CREATE OR
REPLACE TEMPORARY FUNCTION f` again with the same class makes `SELECT * FROM v`
fail with `INCOMPATIBLE_VIEW_SCHEMA_CHANGE`.
The comment at L2092 says column names use the function name, which doesn't
hold for magic methods. Could an `Invoke` / `StaticInvoke` carrying FUNC_ALIAS
get its column name from the function name too, with a column-name assertion
for a magic method? If you'd rather keep the current names, the comment should
be narrowed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2167,12 +2228,58 @@ 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 V2 function reports its own description, which requires instantiating
it. Otherwise, or if
+ * the class cannot be loaded or instantiated, use the class-name-only info,
so that like a Hive
+ * UDF, a broken function only fails when invoked. The class of a function
with `USING`
+ * resources is not touched, as initializing it may require those resources,
which are only
+ * loaded on first invocation.
+ */
+ private def makeExprInfoForClassFunction(func: CatalogFunction):
ExpressionInfo = {
+ if (func.resources.nonEmpty) {
+ return makeExprInfoForHiveFunction(func)
+ }
+ try {
+ val clazz = Utils.classForName(func.className, initialize = false)
+ if (isV2FunctionClass(clazz)) {
+ makeExprInfoForV2Function(func, newV2FunctionInstance(clazz,
func.identifier))
Review Comment:
For a persistent V2 function without `USING` that this session hasn't
invoked yet, DESCRIBE FUNCTION, `spark.catalog.listFunctions()` and Thrift
`getFunctions` construct an instance here, so the static initializer runs. If
it fails at this point, the JVM marks the class as failed to initialize, and
every later call throws `NoClassDefFoundError: Could not initialize class`
until the driver restarts. For example, if the static initializer uses a class
from a jar added later with `ADD JAR`, a single listing leaves the function
unusable. I reproduced this with a static initializer that throws until a
system property is set. Hive UDFs only go through `initialize = false` here and
are not affected.
The `USING` branch already leaves the class alone. Could these functions get
the class-name-only info too, with the description showing up after the first
invocation? That would mean narrowing item 2 of the user-facing section in the
PR description and updating the tests that read the description without
invoking the function (`CreateV2FunctionSuite` L188 and L212), plus the
metadata-path `checkFatal` calls in the L386 test. Documenting that metadata
reads initialize the class is the other option. I'd like to hear your view
first.
--
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]
