[SPARK-50698][SQL] Refactor CreateUserDefinedFunction command to extend from UnaryRunnableCommand

project/MimaExcludes.scala

@@ -70,7 +70,30 @@ 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[IncompatibleMethTypeProblem](
+      "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[IncompatibleMethTypeProblem](
+      "org.apache.spark.sql.execution.command.CreateSQLFunctionCommand.apply"),
+    ProblemFilters.exclude[IncompatibleMethTypeProblem](
+      "org.apache.spark.sql.execution.command.CreateSQLFunctionCommand.copy"),
+    ProblemFilters.exclude[IncompatibleMethTypeProblem](
+      "org.apache.spark.sql.execution.command.CreateSQLFunctionCommand.this"),
+    ProblemFilters.exclude[IncompatibleResultTypeProblem](
+      "org.apache.spark.sql.execution.command.CreateSQLFunctionCommand.copy$default$1"),
+    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

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(

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) =>

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,14 +1642,16 @@ case class CreateUserDefinedFunction(
     collation: Option[String],
     isDeterministic: Option[Boolean],
     containsSQL: Option[Boolean],
-    language: RoutineLanguage,
+    language: org.apache.spark.sql.catalyst.catalog.RoutineLanguage,
     isTableFunc: Boolean,
+    isTemp: 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.
  */

sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala

@@ -698,26 +698,29 @@ class ResolveSessionCatalog(val catalogManager: CatalogManager)
     case CreateFunction(ResolvedIdentifier(catalog, _), _, _, _, _) =>
       throw QueryCompilationErrors.missingCatalogCreateFunctionAbilityError(catalog)
 
-    case c @ CreateUserDefinedFunction(
-        CreateFunctionInSessionCatalog(ident), _, _, _, _, _, _, _, _, _, _, _, _) =>
+    case c @ CreateUserDefinedFunction(child, _, _, _, _, _, _, _, _, _, _, _, _, _)
+        if c.isTemp || (child match {
+          case CreateFunctionInSessionCatalog(_) => true
+          case _ => false
+        }) =>
       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,
-        isTemp = false,
-        c.ignoreIfExists,
-        c.replace)
+        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 = c.isTemp,
+        ignoreIfExists = c.ignoreIfExists,
+        replace = c.replace)
 
     case CreateUserDefinedFunction(
-        ResolvedIdentifier(catalog, _), _, _, _, _, _, _, _, _, _, _, _, _) =>
+        ResolvedIdentifier(catalog, _), _, _, _, _, _, _, _, _, _, _, _, _, _) =>
       throw QueryCompilationErrors.missingCatalogCreateFunctionAbilityError(catalog)
   }
 

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

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkSqlParser.scala

@@ -1053,6 +1053,7 @@ class SparkSqlAstBuilder extends AstBuilder {
             containsSQL,
             language,
             isTableFunc,
+            isTemp = false,
             ctx.EXISTS != null,
             ctx.REPLACE != null)
         } else {
@@ -1063,8 +1064,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,
@@ -1077,8 +1078,7 @@ class SparkSqlAstBuilder extends AstBuilder {
             isTableFunc,
             isTemp = true,
             ctx.EXISTS != null,
-            ctx.REPLACE != null
-          )
+            ctx.REPLACE != null)
         }
       })
     }

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}
@@ -51,7 +51,7 @@ import org.apache.spark.sql.types.{DataType, MetadataBuilder, StructField, Struc
  * }}}
  */
 case class CreateSQLFunctionCommand(
-    name: FunctionIdentifier,
+    child: LogicalPlan,
     inputParamText: Option[String],
     returnTypeText: String,
     exprText: Option[String],
@@ -68,7 +68,35 @@ case class CreateSQLFunctionCommand(
 
   import SQLFunction._
 
+  lazy val name: FunctionIdentifier = {
+    val rawIdent = child match {
+      case ResolvedIdentifier(c, ident) =>
+        FunctionIdentifier(ident.name(), ident.namespace().headOption, Some(c.name()))
+      case u: UnresolvedIdentifier =>
+        val parts = u.nameParts
+        if (parts.length >= 3) {
+          FunctionIdentifier(parts.last, Some(parts(parts.length - 2)), Some(parts.head))
+        } else if (parts.length == 2) {
+          FunctionIdentifier(parts.last, Some(parts.head), None)
+        } else {
+          FunctionIdentifier(parts.last, None, None)
+        }
+      case _ =>
+        throw SparkException.internalError(
+          s"Unexpected child plan in CreateSQLFunctionCommand: $child")
+    }
+    if (isTemp) {
+      FunctionIdentifier(rawIdent.funcName, None, None)
+    } else {
+      rawIdent
+    }
+  }
+
+  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
@@ -412,7 +440,7 @@ case class CreateSQLFunctionCommand(
             }
             // Check cyclic reference using qualified function names.
             val newPath = path :+ f.function.name
-            if (f.function.name == name) {
+            if (isSameFunction(f.function.name)) {
               throw UserDefinedFunctionErrors.cyclicFunctionReference(newPath.mkString(" -> "))
             }
             val plan = catalog.makeSQLTableFunctionPlan(f.name, f.function, f.inputs, f.output)
@@ -423,6 +451,14 @@ case class CreateSQLFunctionCommand(
       }
     }
 
+    def isSameFunction(fName: FunctionIdentifier): Boolean = {
+      if (isTemp) {
+        fName.funcName == name.funcName
+      } else {
+        fName == name
+      }
+    }
+
     def checkExpression(expression: Expression, path: Seq[FunctionIdentifier]): Unit = {
       expression.foreach {
         case s: SubqueryExpression => checkPlan(s.plan, path)
@@ -435,7 +471,7 @@ case class CreateSQLFunctionCommand(
             }
             // Check cyclic reference using qualified function names.
             val newPath = path :+ f.function.name
-            if (f.function.name == name) {
+            if (isSameFunction(f.function.name)) {
               throw UserDefinedFunctionErrors.cyclicFunctionReference(newPath.mkString(" -> "))
             }
             val plan = catalog.makeSQLFunctionPlan(f.name, f.function, f.inputs)

sql/core/src/main/scala/org/apache/spark/sql/execution/command/CreateUserDefinedFunctionCommand.scala

@@ -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 {
@@ -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.

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,8 +59,9 @@ class CreateSQLFunctionParserSuite extends AnalysisTest {
       collation = None,
       isDeterministic = isDeterministic,
       containsSQL = containsSQL,
-      language = LanguageSQL,
+      language = org.apache.spark.sql.catalyst.catalog.LanguageSQL,
       isTableFunc = isTableFunc,
+      isTemp = false,
       ignoreIfExists = ignoreIfExists,
       replace = replace)
   }
@@ -79,10 +78,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,6 +90,7 @@ class CreateSQLFunctionParserSuite extends AnalysisTest {
       collation = None,
       isDeterministic = isDeterministic,
       containsSQL = containsSQL,
+      language = org.apache.spark.sql.catalyst.catalog.LanguageSQL,
       isTableFunc = isTableFunc,
       isTemp = true,
       ignoreIfExists = ignoreIfExists,