dongjoon-hyun commented on code in PR #59697:
URL: https://github.com/apache/spark/pull/59697#discussion_r4371949209
##########
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:
**A `description()` that throws now fails `CREATE TEMPORARY FUNCTION`,
`REFRESH FUNCTION` and every first call of a persistent function, although the
function itself works.** Since the last round, registration builds its info
here from the bound instance, and `unbound.description()` at L2262 runs outside
any `try`. Before that change, registration went through
`makeExprInfoForClassFunction`, whose `case NonFatal(_) | _: LinkageError` at
L2251 falls back to the class-name-only info. Take a class whose `bind` and
`produceResult` work but whose `description()` is `???` (a
`NotImplementedError`, which is `NonFatal`), a stub that throws
`UnsupportedOperationException`, or one that hits an NPE:
- `CREATE TEMPORARY FUNCTION f AS '...'` fails with the raw exception and no
error class, and an NPE is reported as `INTERNAL_ERROR`.
- `CREATE FUNCTION f AS '...'` succeeds, and SHOW, DESCRIBE and
`listFunctions` work through the fallback at L2251. But every `SELECT f(...)`
fails in the builder factory at L2835, and since nothing gets registered, the
next query fails again.
- The same class works in queries when a `FunctionCatalog` serves it,
because `resolveV2Function` never calls `description()`.
sunchao's review summary also relied on "Description failures fall back to
basic metadata, while invocation retains its normal failure behavior." I traced
this by reading the code and did not run it. Could this keep the reused
instance but guard the description, e.g. `try makeExprInfoForV2Function(func,
unbound) catch { case NonFatal(_) => makeExprInfoForHiveFunction(func) }`, and
add a test with a throwing `description()`?
##########
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:
Minor: errors raised inside `bindFunction` still name the V2 object instead
of the registered function, because the alias is set only after it returns
(L2097). For example, after `CREATE TEMPORARY FUNCTION my_len AS
'org.apache.spark.sql.connector.V2StrLen'`, whose `name()` is `strlen`, `SELECT
my_len(1)` reports "Function 'strlen' cannot process input: (int): Expect
StringType.", while `SELECT my_len(DISTINCT 'a')` reports `my_len`.
`_LEGACY_ERROR_TEMP_1199` also uses the bound function's `name()`, and
`INVALID_UDF_IMPLEMENTATION` and `SCALAR_FUNCTION_NOT_FULLY_IMPLEMENTED`
(V2ExpressionUtils.scala:245 and :277) name the bound function, e.g.
`strlen_default`. No test checks a bind-time error for a function registered
under a name other than its `name()`. The `FunctionCatalog` route keeps
`unbound.name()`, which DataSourceV2FunctionSuite.scala:638 pins, so could
`bindFunction` take an optional display name that only this builder passes?
##########
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:
Minor: this arm wraps every cause, including fatal ones.
`Constructor.newInstance` wraps anything the constructor throws in an
`InvocationTargetException`, so an `OutOfMemoryError`, a `StackOverflowError`
or an `InterruptedException` (from a constructor declared `throws
InterruptedException`) becomes `CANNOT_INSTANTIATE_FUNCTION_CLASS`. On the
metadata paths, `case NonFatal(_) | _: LinkageError` at L2251 then catches that
`AnalysisException`, so DESCRIBE and `listFunctions` return normally while the
OOM or the interrupt is lost. cloud-fan's earlier comment asked that "Catching
LinkageError must not also swallow VirtualMachineError or other fatal
throwables; those must keep propagating", and this conversion gets around that
guard. Could this arm wrap only when `NonFatal(e.getCause) ||
e.getCause.isInstanceOf[LinkageError]`, and rethrow `e.getCause` otherwise?
##########
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:
Minor: on the query path, this eager info constructs the class one extra
time. `FunctionResolution` only calls `invoke` on the returned `V1Function`
(FunctionResolution.scala:170-171), and the builder factory then constructs
another instance at L2835. So the first query in a session that calls a
persistent V2 function without `USING` resources runs its constructor and
`description()` twice, and twice per query while the constructor keeps failing.
`spark.catalog.functionExists`, which only checks that the name resolves,
constructs it on every call until the first invocation registers it. This is
what remains of sunchao's construction count after the last round. Could
`V1Function` take its info lazily, e.g. as `() => ExpressionInfo`, so that only
DESCRIBE and the catalog APIs build it and the first call constructs the class
once?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2167,12 +2222,53 @@ 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.
+ * 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 {
+ if (Utils.classIsLoadableAndAssignableFrom(func.className,
classOf[UnboundFunction])) {
Review Comment:
Minor: metadata lookups now touch the class of every class-based function
without `USING` resources on every call, and nothing caches the result. Before
this PR, `lookupPersistentFunction` and the eager info in
`loadPersistentScalarFunction` used `makeExprInfoForHiveFunction`, which never
loads the class. Now:
- every DESCRIBE before first use, `spark.catalog.listFunctions()` and
`getFunction` (classic and Connect), and Thrift `GetFunctions`
(SparkGetFunctionsOperation.scala:83) construct each unused V2 function and
call its `description()`, once per function per call, because only the first
invocation or `REFRESH FUNCTION` registers it;
- a plain Hive UDF whose class is not on Spark's classpath (e.g. one that
relies on `hive.aux.jars.path` and has no resource URIs) gets a failed
`Class.forName` and a new `ClassNotFoundException` on every lookup;
- each check looks the class up twice, in `classIsLoadableAndAssignableFrom`
and again in `Utils.classForName` at L2245.
With N such functions, one listing does N constructions and 2N lookups, and
every Thrift or Connect session repeats it. A light constructor disappears next
to the two metastore calls per function, but a heavy one slows down every
listing. I traced this by reading the code and did not run it. Could the V2
info be memoized per session, keyed by the identifier and the class name and
dropped on DROP, ALTER, REFRESH and CREATE OR REPLACE, and could L2244-2245
become one `Utils.classForName(func.className, initialize = false)` followed by
`isV2FunctionClass(clazz)` inside the same `try`?
##########
docs/sql-migration-guide.md:
##########
@@ -59,6 +59,7 @@ license: |
- Since Spark 4.4, with whole-stage codegen enabled, a filter condition,
including the `FILTER` clause of an aggregate function, evaluates the remaining
top-level conjuncts of the condition on a row after one of them evaluates to
`NULL`, as the SQL standard requires, and does so in the order they appear in
the condition, as Spark already did with whole-stage codegen disabled, which
also evaluates a subexpression shared by several conjuncts before the
condition. Only a `FALSE` conjunct stops the evaluation. One exception: an `IS
NOT NULL` check on a column, or on an expression over it, that a later conjunct
references is still evaluated before that conjunct and the conjuncts up to the
check's own position, even when it is written after them, and can drop the row
before an error one of them raises. Once a condition has a conjunct that can be
`NULL`, the `IS NOT NULL` checks after that conjunct are evaluated in the order
of the condition on every row, not only on rows where a conjunct
is `NULL`. A row whose condition is `NULL` is still dropped, but an error
raised by a later conjunct on that row, such as `ARITHMETIC_OVERFLOW` in ANSI
mode, now fails the query. For example, `SELECT a FROM t WHERE NOT (a > 0 AND b
> 0) AND y + 1 > 0` can fail with `ARITHMETIC_OVERFLOW` on a row where `a` is
`NULL`, `b` is `1`, and `y` is `2147483647`, unless the row is dropped earlier
by a `FALSE` conjunct or `IS NOT NULL` check that precedes `y + 1 > 0` in the
condition, or skipped before the filter, for example by data source filter
pushdown such as Parquet row group statistics. Previously the query skipped
such a row with whole-stage codegen enabled. The later conjuncts also run on
more rows than before, so a nondeterministic conjunct such as `rand(seed)` can
return different values on the remaining rows, and a side-effecting
user-defined function is called on rows that are dropped. To restore the
previous behavior, set `spark.sql.legacy.filterShortCircuitOnNullConjunct` to
`tru
e`.
- Since Spark 4.4, when a format is specified, `to_timestamp`,
`try_to_timestamp`, and `to_timestamp_ltz` cast a numeric first argument to a
string and parse it with the format, for example `to_timestamp(20201231,
'yyyyMMdd')` returns `2020-12-31 00:00:00`. This matches the existing behavior
when `spark.sql.timestampType` is `TIMESTAMP_NTZ`. Previously, with the default
`TIMESTAMP_LTZ` timestamp type, such calls failed with `INTERNAL_ERROR`. Note
that the string form of the number is parsed, so a DECIMAL `1.5` is parsed as
`"1.5"`.
+- Since Spark 4.4, a function created by `CREATE FUNCTION ... AS 'class'`
whose class both extends a Hive UDF class (such as `UDF` or `GenericUDF`) and
implements `org.apache.spark.sql.connector.catalog.functions.UnboundFunction`
runs as a DataSource V2 function instead of a Hive UDF, including functions
created in earlier versions. Its result type, argument coercion, and null
handling are now defined by `UnboundFunction.bind` instead of the Hive UDF.
Review Comment:
Minor: this note covers only part of the change. The V2 check at
SessionCatalog.scala:2087 runs before `functionExpressionBuilder`, so:
- a class that extends
`org.apache.spark.sql.expressions.UserDefinedAggregateFunction` (handled by
`SparkUDFExpressionBuilder` at BaseSessionStateBuilder.scala:503) and
implements `UnboundFunction` also switches, from `ScalaUDAF` to `V2Aggregator`;
- determinism and constant folding change too. A Hive wrapper is
deterministic only when its `@UDFType` says so (hiveUDFEvaluators.scala:46-48),
and a `GenericUDF` call is foldable only with a constant return inspector
(hiveUDFs.scala:141-142). As a V2 function it uses
`BoundFunction.isDeterministic()`, which defaults to `true`, and
`ApplyFunctionExpression` is foldable whenever its arguments are
(ApplyFunctionExpression.scala:35-37). So a `@UDFType(deterministic = false)`
class that does not override `isDeterministic()` becomes deterministic, and
`f('k')` can be folded once on the driver during planning instead of running
per row.
Could the note say "extends a Hive UDF class (such as `UDF` or `GenericUDF`)
or `UserDefinedAggregateFunction`", and add determinism to the semantics that
`UnboundFunction.bind` now defines?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionResolution.scala:
##########
@@ -796,80 +792,41 @@ class FunctionResolution(
unbound: UnboundFunction,
arguments: Seq[Expression],
unresolvedFunc: UnresolvedFunction): Expression = {
- val inputType = StructType(arguments.zipWithIndex.map {
- case (exp, pos) => StructField(s"_$pos", exp.dataType, exp.nullable)
- })
- val bound = try {
- unbound.bind(inputType)
- } catch {
- case unsupported: UnsupportedOperationException =>
- throw QueryCompilationErrors.functionCannotProcessInputError(
- unbound,
- arguments,
- unsupported
- )
- }
-
- if (bound.inputTypes().length != arguments.length) {
- throw
QueryCompilationErrors.v2FunctionInvalidInputTypeLengthError(bound, arguments)
- }
-
- bound match {
- case scalarFunc: ScalarFunction[_] =>
- processV2ScalarFunction(scalarFunc, arguments, unresolvedFunc)
- case aggFunc: V2AggregateFunction[_, _] =>
- processV2AggregateFunction(aggFunc, arguments, unresolvedFunc)
- case _ =>
- failAnalysis(
- errorClass = "INVALID_UDF_IMPLEMENTATION",
- messageParameters = Map("funcName" -> toSQLId(bound.name()))
- )
+ V2ExpressionUtils.bindFunction(unbound, arguments) match {
+ case aggregator: V2Aggregator[_, _] =>
+ processV2AggregateFunction(aggregator, unresolvedFunc)
+ case scalarExpr =>
+ checkV2ScalarFunctionSyntax(unbound.name(), unresolvedFunc)
+ scalarExpr
}
}
- private def processV2ScalarFunction(
- scalarFunc: ScalarFunction[_],
- arguments: Seq[Expression],
- unresolvedFunc: UnresolvedFunction): Expression = {
+ private def checkV2ScalarFunctionSyntax(
Review Comment:
Nit: after the split, this is the same check as
`checkUnsupportedAggregateClause` at L732-751: the same three conditions in the
same order, with the same syntax strings, and only the name parameter differs
(a `String` here, `func.prettyName` there). Every branch throws, so the `else
if` chain and the independent `if`s behave the same. Could
`checkUnsupportedAggregateClause` take the name as a `String`, with L685 and
L717 passing `agg.prettyName` and `other.prettyName` and L799 passing
`unbound.name()`? The error names stay exactly as they are now.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala:
##########
@@ -2736,12 +2832,9 @@ class SessionCatalog(
makeUserDefinedScalarFuncBuilder(udf))
} else {
loadFunctionResources(funcMetadata.resources)
+ val (builder, loadedInfo) =
makeFunctionBuilderAndInfo(funcMetadata)
Review Comment:
Nit: these lines repeat the `case None` branch of the public
`registerFunction` at L2175-2177, which builds the builder and the info from
one instance and registers them with `Some(info)` in the same
`functionRegistry`. Could L2834-2837 be
`loadFunctionResources(funcMetadata.resources)` followed by
`registerFunction(funcMetadata, overrideIfExists = false)`? The two-argument
call can only bind to the public overload, the next change to class-function
registration would then be made in one place, and `loadedInfo`, which only
avoids shadowing `info` at L2816, goes away.
--
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]
