Skip to content
Open
11 changes: 10 additions & 1 deletion project/MimaExcludes.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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) =>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -1642,14 +1642,15 @@ 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 {
override protected def withNewChildInternal(newChild: LogicalPlan): CreateUserDefinedFunction =
copy(child = newChild)
}


/**
* The logical plan of the DROP FUNCTION command.
*/
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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, _), _, _, _, _, _, _, _, _, _, _, _, _) =>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -1075,10 +1075,8 @@ class SparkSqlAstBuilder extends AstBuilder {
containsSQL,
language,
isTableFunc,
isTemp = true,
ctx.EXISTS != null,
ctx.REPLACE != null
)
ctx.REPLACE != null)
}
})
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand All @@ -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],
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,17 @@ 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

/**
* The base class for CreateUserDefinedFunctionCommand
*/
abstract class CreateUserDefinedFunctionCommand
extends LeafRunnableCommand with CapturesConfig
extends UnaryRunnableCommand with CapturesConfig


object CreateUserDefinedFunctionCommand {
Expand All @@ -40,7 +42,7 @@ object CreateUserDefinedFunctionCommand {
*/
// scalastyle:off argcount
def apply(
name: FunctionIdentifier,
child: LogicalPlan,
inputParamText: Option[String],
returnTypeText: String,
exprText: Option[String],
Expand All @@ -62,7 +64,7 @@ object CreateUserDefinedFunctionCommand {
language match {
case LanguageSQL =>
CreateSQLFunctionCommand(
name,
child,
inputParamText,
returnTypeText,
exprText,
Expand All @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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)
Expand All @@ -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,
Expand All @@ -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)
}
Expand Down