pan3793 commented on code in PR #59697:
URL: https://github.com/apache/spark/pull/59697#discussion_r4374767631
##########
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:
@dongjoon-hyun Merged the two lookups into one `classForName(..., initialize
= false)` in 45f1d6bdf84. I'd rather not cache the info: it's only built before
the function's first invocation in a session, where each lookup already makes
metastore calls, so a cache wouldn't pay for its invalidation logic.
##########
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:
@dongjoon-hyun Updated in 45f1d6bdf84 to cover
`UserDefinedAggregateFunction`, and determinism and constant folding.
##########
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:
@dongjoon-hyun Done in 45f1d6bdf84.
##########
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:
@dongjoon-hyun Done in 45f1d6bdf84.
--
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]
