diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index 3701646cf7cde..b0dd69843c97f 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -63,7 +63,16 @@ object MimaExcludes { // [SPARK-57987] Add desc field to the SQL REST API Node case class ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.apply"), ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.copy"), - ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.sql.Node$") + ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.sql.Node$"), + // [SPARK-50698][SQL] Refactor CreateUserDefinedFunctionCommand to extend from UnaryRunnableCommand + ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.sql.execution.command.CreateUserDefinedFunctionCommand"), + ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.execution.command.CreateUserDefinedFunctionCommand.apply"), + ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.sql.execution.command.CreateSQLFunctionCommand"), + ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.sql.execution.command.CreateSQLFunctionCommand.apply"), + ProblemFilters.exclude[MissingClassProblem]( + "org.apache.spark.sql.catalyst.plans.logical.CreateUserDefinedFunction"), + ProblemFilters.exclude[MissingClassProblem]( + "org.apache.spark.sql.catalyst.plans.logical.CreateUserDefinedFunction$") ) // Exclude rules for 4.2.x from 4.1.0 diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyDefaultCollation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyDefaultCollation.scala index ff43b3668839b..103dd45ed3062 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyDefaultCollation.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyDefaultCollation.scala @@ -206,9 +206,9 @@ object ApplyDefaultCollation extends Rule[LogicalPlan] { newCreateView.copyTagsFrom(createView) newCreateView - case createUserDefinedFunction@CreateUserDefinedFunction( - ResolvedIdentifier(catalog: SupportsNamespaces, identifier), - _, _, _, _, _, collation, _, _, _, _, _, _) if collation.isEmpty => + case createUserDefinedFunction@CreateUserDefinedFunction(ResolvedIdentifier( + catalog: SupportsNamespaces, identifier), _, _, _, _, _, _, _, _, _, _, _, _) + if createUserDefinedFunction.collation.isEmpty => val newCreateUserDefinedFunction = CurrentOrigin.withOrigin(createUserDefinedFunction.origin) { createUserDefinedFunction.copy( diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala index 6fc196774a048..45001dad6b08c 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala @@ -91,11 +91,13 @@ class ResolveCatalogs(val catalogManager: CatalogManager) throw QueryCompilationErrors.operationNotAllowedOnBuiltinFunctionError( "CREATE", nameParts.last) - case CreateUserDefinedFunction(UnresolvedIdentifier(nameParts, _), - _, _, _, _, _, _, _, _, _, _, _, _) - if isSystemBuiltinName(nameParts) => - throw QueryCompilationErrors.operationNotAllowedOnBuiltinFunctionError( - "CREATE", nameParts.last) + case c @ CreateUserDefinedFunction( + u @ UnresolvedIdentifier(nameParts, _), _, _, _, _, _, _, _, _, _, _, _, _) => + if (isSystemBuiltinName(nameParts)) { + throw QueryCompilationErrors.operationNotAllowedOnBuiltinFunctionError( + "CREATE", nameParts.last) + } + c.copy(child = resolveFunctionIdentifier(nameParts, u.origin)) case DropFunction(UnresolvedIdentifier(nameParts, _), _) if isSystemBuiltinName(nameParts) => diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala index b816016a3ec84..32ab97c94ecf1 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2Commands.scala @@ -21,8 +21,8 @@ import org.apache.spark.{SparkException, SparkIllegalArgumentException, SparkUns import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.catalyst.analysis.{AnalysisContext, AssignmentUtils, EliminateSubqueryAliases, FieldName, NamedRelation, PartitionSpec, ResolvedIdentifier, ResolvedProcedure, ResolveSchemaEvolution, TypeCheckResult, UnresolvedAttribute, UnresolvedException, UnresolvedProcedure, ViewSchemaMode} import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess} -import org.apache.spark.sql.catalyst.catalog.{FunctionResource, RoutineLanguage} import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec +import org.apache.spark.sql.catalyst.catalog.FunctionResource import org.apache.spark.sql.catalyst.expressions._ import org.apache.spark.sql.catalyst.plans.DescribeCommandSchema import org.apache.spark.sql.catalyst.trees.BinaryLike @@ -1642,7 +1642,7 @@ case class CreateUserDefinedFunction( collation: Option[String], isDeterministic: Option[Boolean], containsSQL: Option[Boolean], - language: RoutineLanguage, + language: org.apache.spark.sql.catalyst.catalog.RoutineLanguage, isTableFunc: Boolean, ignoreIfExists: Boolean, replace: Boolean) extends UnaryCommand { @@ -1650,6 +1650,7 @@ case class CreateUserDefinedFunction( copy(child = newChild) } + /** * The logical plan of the DROP FUNCTION command. */ diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala index e6fc6d8d862ce..fe52e4613b7e3 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala @@ -701,20 +701,20 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager) case c @ CreateUserDefinedFunction( CreateFunctionInSessionCatalog(ident), _, _, _, _, _, _, _, _, _, _, _, _) => CreateUserDefinedFunctionCommand( - FunctionIdentifier(ident.table, ident.database, ident.catalog), - c.inputParamText, - c.returnTypeText, - c.exprText, - c.queryText, - c.comment, - c.collation, - c.isDeterministic, - c.containsSQL, - c.language, - c.isTableFunc, + child = c.child, + inputParamText = c.inputParamText, + returnTypeText = c.returnTypeText, + exprText = c.exprText, + queryText = c.queryText, + comment = c.comment, + collation = c.collation, + isDeterministic = c.isDeterministic, + containsSQL = c.containsSQL, + language = c.language, + isTableFunc = c.isTableFunc, isTemp = false, - c.ignoreIfExists, - c.replace) + ignoreIfExists = c.ignoreIfExists, + replace = c.replace) case CreateUserDefinedFunction( ResolvedIdentifier(catalog, _), _, _, _, _, _, _, _, _, _, _, _, _) => diff --git a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala index cb2fae1d9af9c..5effb01f13d9f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/catalyst/parser/SqlStatementCodes.scala @@ -146,8 +146,7 @@ object SqlStatementCodes { case _: SetCatalogAndNamespace | _: SetNamespaceCommand => SetSchema case _: SetCatalogCommand => SetCatalog case _: TruncateTable => TruncateTable - case _: CreateFunction | _: CreateFunctionCommand | - _: CreateUserDefinedFunction | _: CreateUserDefinedFunctionCommand => + case _: CreateFunction | _: CreateFunctionCommand | _: CreateUserDefinedFunctionCommand => CreateRoutine case _: DropFunction | _: DropFunctionCommand => DropRoutine case _: UnresolvedExecuteImmediate => ExecuteImmediate diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala index 5070a40259e37..3a7e0bc5b3087 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala @@ -1063,8 +1063,8 @@ class SparkSqlAstBuilder extends AstBuilder { // Extract the actual function name, handling session qualification val funcName = extractTempFunctionName(functionIdentifier, ctx) - CreateUserDefinedFunctionCommand( - FunctionIdentifier(funcName), + CreateUserDefinedFunction( + UnresolvedIdentifier(Seq(funcName)), inputParamText, returnTypeText, exprText, @@ -1075,10 +1075,8 @@ class SparkSqlAstBuilder extends AstBuilder { containsSQL, language, isTableFunc, - isTemp = true, ctx.EXISTS != null, - ctx.REPLACE != null - ) + ctx.REPLACE != null) } }) } diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionCommand.scala index a597087085b42..809123f87749b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionCommand.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionCommand.scala @@ -20,7 +20,7 @@ package org.apache.spark.sql.execution.command import org.apache.spark.SparkException import org.apache.spark.sql.{AnalysisException, Row, SparkSession} import org.apache.spark.sql.catalyst.FunctionIdentifier -import org.apache.spark.sql.catalyst.analysis.{withPosition, Analyzer, SQLFunctionExpression, SQLFunctionNode, SQLScalarFunction, SQLTableFunction, UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedTableValuedFunction} +import org.apache.spark.sql.catalyst.analysis.{withPosition, Analyzer, ResolvedIdentifier, SQLFunctionExpression, SQLFunctionNode, SQLScalarFunction, SQLTableFunction, UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedIdentifier, UnresolvedRelation, UnresolvedTableValuedFunction} import org.apache.spark.sql.catalyst.catalog.{SessionCatalog, SQLFunction, UserDefinedFunction, UserDefinedFunctionErrors} import org.apache.spark.sql.catalyst.catalog.UserDefinedFunction._ import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, Expression, Generator, LateralSubquery, Literal, ScalarSubquery, SubqueryExpression, WindowExpression} @@ -34,24 +34,8 @@ import org.apache.spark.sql.errors.QueryCompilationErrors import org.apache.spark.sql.execution.command.CreateUserDefinedFunctionCommand._ import org.apache.spark.sql.types.{DataType, MetadataBuilder, StructField, StructType} -/** - * The DDL command that creates a SQL function. - * For example: - * {{{ - * CREATE [OR REPLACE] [TEMPORARY] FUNCTION [IF NOT EXISTS] [db_name.]function_name - * ([param_name param_type [COMMENT param_comment], ...]) - * RETURNS {ret_type | TABLE (ret_name ret_type [COMMENT ret_comment], ...])} - * [function_properties] function_body; - * - * function_properties: - * [NOT] DETERMINISTIC | COMMENT function_comment | [ CONTAINS SQL | READS SQL DATA ] - * - * function_body: - * RETURN {expression | TABLE ( query )} - * }}} - */ case class CreateSQLFunctionCommand( - name: FunctionIdentifier, + child: LogicalPlan, inputParamText: Option[String], returnTypeText: String, exprText: Option[String], @@ -68,7 +52,21 @@ case class CreateSQLFunctionCommand( import SQLFunction._ + lazy val name: FunctionIdentifier = child match { + case ResolvedIdentifier(c, ident) => + FunctionIdentifier(ident.name(), ident.namespace().headOption) + case u: UnresolvedIdentifier => + FunctionIdentifier(u.nameParts.last, u.nameParts.dropRight(1).lastOption) + case _ => + throw SparkException.internalError( + s"Unexpected child plan in CreateSQLFunctionCommand: $child") + } + + override protected def withNewChildInternal( + newChild: LogicalPlan): CreateSQLFunctionCommand = copy(child = newChild) + override def run(sparkSession: SparkSession): Seq[Row] = { + val parser = sparkSession.sessionState.sqlParser val analyzer = sparkSession.sessionState.analyzer val catalog = sparkSession.sessionState.catalog diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateUserDefinedFunctionCommand.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateUserDefinedFunctionCommand.scala index f65c7c91251a1..a73d01e1c9145 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateUserDefinedFunctionCommand.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateUserDefinedFunctionCommand.scala @@ -21,7 +21,9 @@ import java.util.Locale import org.apache.spark.sql.AnalysisException import org.apache.spark.sql.catalyst.{CapturesConfig, FunctionIdentifier} +import org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier import org.apache.spark.sql.catalyst.catalog.{LanguageSQL, RoutineLanguage, UserDefinedFunctionErrors} +import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.StructType @@ -29,7 +31,7 @@ import org.apache.spark.sql.types.StructType * The base class for CreateUserDefinedFunctionCommand */ abstract class CreateUserDefinedFunctionCommand - extends LeafRunnableCommand with CapturesConfig + extends UnaryRunnableCommand with CapturesConfig object CreateUserDefinedFunctionCommand { @@ -40,7 +42,7 @@ object CreateUserDefinedFunctionCommand { */ // scalastyle:off argcount def apply( - name: FunctionIdentifier, + child: LogicalPlan, inputParamText: Option[String], returnTypeText: String, exprText: Option[String], @@ -62,7 +64,7 @@ object CreateUserDefinedFunctionCommand { language match { case LanguageSQL => CreateSQLFunctionCommand( - name, + child, inputParamText, returnTypeText, exprText, @@ -80,6 +82,42 @@ object CreateUserDefinedFunctionCommand { throw UserDefinedFunctionErrors.unsupportedUserDefinedFunction(other) } } + // scalastyle:off argcount + def apply( + name: FunctionIdentifier, + inputParamText: Option[String], + returnTypeText: String, + exprText: Option[String], + queryText: Option[String], + comment: Option[String], + collation: Option[String], + isDeterministic: Option[Boolean], + containsSQL: Option[Boolean], + language: RoutineLanguage, + isTableFunc: Boolean, + isTemp: Boolean, + ignoreIfExists: Boolean, + replace: Boolean + ): CreateUserDefinedFunctionCommand = { + // scalastyle:on argcount + val nameParts = name.database.toSeq :+ name.funcName + apply( + UnresolvedIdentifier(nameParts), + inputParamText, + returnTypeText, + exprText, + queryText, + comment, + collation, + isDeterministic, + containsSQL, + language, + isTableFunc, + isTemp, + ignoreIfExists, + replace) + } + /** * Check whether the function parameters contain duplicated column names. diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionParserSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionParserSuite.scala index 56316f43f8dfe..c6e7992e7d60d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionParserSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/CreateSQLFunctionParserSuite.scala @@ -18,9 +18,7 @@ package org.apache.spark.sql.execution.command import org.apache.spark.sql.AnalysisException -import org.apache.spark.sql.catalyst.FunctionIdentifier import org.apache.spark.sql.catalyst.analysis.{AnalysisTest, UnresolvedIdentifier} -import org.apache.spark.sql.catalyst.catalog.LanguageSQL import org.apache.spark.sql.catalyst.plans.logical.CreateUserDefinedFunction import org.apache.spark.sql.execution.SparkSqlParser @@ -61,7 +59,7 @@ class CreateSQLFunctionParserSuite extends AnalysisTest { collation = None, isDeterministic = isDeterministic, containsSQL = containsSQL, - language = LanguageSQL, + language = org.apache.spark.sql.catalyst.catalog.LanguageSQL, isTableFunc = isTableFunc, ignoreIfExists = ignoreIfExists, replace = replace) @@ -79,10 +77,10 @@ class CreateSQLFunctionParserSuite extends AnalysisTest { containsSQL: Option[Boolean] = None, isTableFunc: Boolean = false, ignoreIfExists: Boolean = false, - replace: Boolean = false): CreateSQLFunctionCommand = { + replace: Boolean = false): CreateUserDefinedFunction = { // scalastyle:on argcount - CreateSQLFunctionCommand( - FunctionIdentifier(name), + CreateUserDefinedFunction( + UnresolvedIdentifier(Seq(name)), inputParamText = inputParamText, returnTypeText = returnTypeText, exprText = exprText, @@ -91,8 +89,8 @@ class CreateSQLFunctionParserSuite extends AnalysisTest { collation = None, isDeterministic = isDeterministic, containsSQL = containsSQL, + language = org.apache.spark.sql.catalyst.catalog.LanguageSQL, isTableFunc = isTableFunc, - isTemp = true, ignoreIfExists = ignoreIfExists, replace = replace) }