[SPARK-51439][SQL] Support SQL UDF with DEFAULT argument

Continuing allisonwang-db's work on #50373 and #49471

This PR adds support for DEFAULT arguments in SQL UDF. Examples:
```sql
CREATE FUNCTION foo1d1(a INT DEFAULT 10) RETURNS INT RETURN a;
SELECT foo1d1();   -- 10
SELECT foo1d1(20); -- 20

CREATE FUNCTION foo1d6(a INT, b INT DEFAULT 7) RETURNS TABLE(a INT, b INT) RETURN SELECT a, b;
SELECT * FROM foo1d6(5);    -- 5, 7
SELECT * FROM foo1d6(5, 2); -- 5, 2
```

See sql-udf.sql for more valid and invalid examples.

To support default arguments in SQL UDFs.

Yes. Now SQL UDFs support DEFAULT arguments.

A side effect of the grammar change is that some invalid function parameter definitions are now no longer rejected by the grammar, but instead rejected by the parser logic.

Examples:

```sql
-- multiple COMMENT or multiple NOT NULL
CREATE TEMPORARY FUNCTION foo(a INT COMMENT 'hello' COMMENT 'world') RETURNS INT RETURN a;

-- before:
[PARSE_SYNTAX_ERROR] Syntax error at or near 'COMMENT'. SQLSTATE: 42601
== SQL (line 2, position 1) ==
CREATE TEMPORARY FUNCTION foo(a INT COMMENT 'hello' COMMENT 'world') RETURNS INT RETURN a;
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

-- after:
[CREATE_TABLE_COLUMN_DESCRIPTOR_DUPLICATE] CREATE TABLE column a specifies descriptor "COMMENT" more than once, which is invalid. SQLSTATE: 42710
== SQL (line 1, position 1) ==
CREATE TEMPORARY FUNCTION foo(a INT COMMENT 'hello' COMMENT 'world')...
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
```

```sql
-- GENERATED ALWAYS AS
CREATE TEMPORARY FUNCTION foo(a INT GENERATED ALWAYS AS (1)) RETURNS INT RETURN a;

-- before:
[PARSE_SYNTAX_ERROR] Syntax error at or near 'GENERATED'. SQLSTATE: 42601
== SQL (line 2, position 1) ==
CREATE TEMPORARY FUNCTION foo(a INT GENERATED ALWAYS AS (1)) RETURNS INT RETURN a;
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

-- after:
[INVALID_SQL_SYNTAX.CREATE_FUNC_WITH_GENERATED_COLUMNS_AS_PARAMETERS] Invalid SQL syntax: CREATE FUNCTION with generated columns as parameters is not allowed. SQLSTATE: 42000
== SQL (line 2, position 1) ==
CREATE TEMPORARY FUNCTION foo(a INT GENERATED ALWAYS AS (1)) RETURNS INT RETURN a;
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
```

This doesn't change the behavior of existing valid SQL.

End-to-end regression tests in `sql-udf.sql` and simple tests in `SQLFunctionSuite`.

No

Closes #50408 from wengh/sql-udf-default.

Lead-authored-by: Haoyu Weng <[email protected]>
Co-authored-by: Allison Wang <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: 2 people

common/utils/src/main/resources/error/error-conditions.json

@@ -3366,6 +3366,16 @@
           "ANALYZE TABLE(S) ... COMPUTE STATISTICS ... <ctx> must be either NOSCAN or empty."
         ]
       },
+      "CREATE_FUNC_WITH_COLUMN_CONSTRAINTS" : {
+        "message" : [
+          "CREATE FUNCTION with constraints on parameters is not allowed."
+        ]
+      },
+      "CREATE_FUNC_WITH_GENERATED_COLUMNS_AS_PARAMETERS" : {
+        "message" : [
+          "CREATE FUNCTION with generated columns as parameters is not allowed."
+        ]
+      },
       "CREATE_ROUTINE_WITH_IF_NOT_EXISTS_AND_REPLACE" : {
         "message" : [
           "Cannot create a routine with both IF NOT EXISTS and REPLACE specified."

sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4

@@ -177,6 +177,10 @@ singleTableSchema
     : colTypeList EOF
     ;
 
+singleRoutineParamList
+    : colDefinitionList EOF
+    ;
+
 statement
     : query                                                            #statementDefault
     | executeImmediate                                                 #visitExecuteImmediate

sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkParserUtils.scala

@@ -127,6 +127,14 @@ trait SparkParserUtils {
     }
   }
 
+  /** Get the code that creates the given node. */
+  def source(ctx: ParserRuleContext): String = {
+    // Note: `exprCtx.getText` returns a string without spaces, so we need to
+    // get the text from the underlying char stream instead.
+    val stream = ctx.getStart.getInputStream
+    stream.getText(Interval.of(ctx.getStart.getStartIndex, ctx.getStop.getStopIndex))
+  }
+
   /** Convert a string token into a string. */
   def string(token: Token): String = unescapeSQLString(token.getText)
 

sql/api/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala

@@ -656,6 +656,16 @@ private[sql] object QueryParsingErrors extends DataTypeErrorsBase {
       ctx)
   }
 
+  def createFuncWithGeneratedColumnsError(ctx: ParserRuleContext): Throwable = {
+    new ParseException(
+      errorClass = "INVALID_SQL_SYNTAX.CREATE_FUNC_WITH_GENERATED_COLUMNS_AS_PARAMETERS",
+      ctx)
+  }
+
+  def createFuncWithConstraintError(ctx: ParserRuleContext): Throwable = {
+    new ParseException(errorClass = "INVALID_SQL_SYNTAX.CREATE_FUNC_WITH_COLUMN_CONSTRAINTS", ctx)
+  }
+
   def defineTempFuncWithIfNotExistsError(ctx: ParserRuleContext): Throwable = {
     new ParseException(errorClass = "INVALID_SQL_SYNTAX.CREATE_TEMP_FUNC_WITH_IF_NOT_EXISTS", ctx)
   }

sql/api/src/main/scala/org/apache/spark/sql/types/StructField.scala

@@ -150,10 +150,13 @@ case class StructField(
   /**
    * Return the default value of this StructField. This is used for storing the default value of a
    * function parameter.
+   *
+   * It is present when the field represents a function parameter with a default value, such as
+   * `CREATE FUNCTION f(arg INT DEFAULT 42) RETURN ...`.
    */
   private[sql] def getDefault(): Option[String] = {
-    if (metadata.contains("default")) {
-      Option(metadata.getString("default"))
+    if (metadata.contains(StructType.SQL_FUNCTION_DEFAULT_METADATA_KEY)) {
+      Option(metadata.getString(StructType.SQL_FUNCTION_DEFAULT_METADATA_KEY))
     } else {
       None
     }
@@ -183,6 +186,9 @@ case class StructField(
 
   /**
    * Return the current default value of this StructField.
+   *
+   * It is present only when the field represents a table column with a default value, such as:
+   * `ALTER TABLE t ALTER COLUMN c SET DEFAULT 42`.
    */
   def getCurrentDefaultValue(): Option[String] = {
     if (metadata.contains(CURRENT_DEFAULT_COLUMN_METADATA_KEY)) {
@@ -214,7 +220,8 @@ case class StructField(
     }
   }
 
-  private def getDDLDefault = getCurrentDefaultValue()
+  private def getDDLDefault = getDefault()
+    .orElse(getCurrentDefaultValue())
     .map(" DEFAULT " + _)
     .getOrElse("")
 

sql/api/src/main/scala/org/apache/spark/sql/types/StructType.scala

@@ -521,6 +521,7 @@ case class StructType(fields: Array[StructField]) extends DataType with Seq[Stru
  */
 @Stable
 object StructType extends AbstractDataType {
+  private[sql] val SQL_FUNCTION_DEFAULT_METADATA_KEY = "default"
 
   override private[sql] def defaultConcreteType: DataType = new StructType
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SQLFunction.scala

@@ -207,7 +207,7 @@ object SQLFunction {
       val returnType = parseReturnTypeText(props(RETURN_TYPE), isTableFunc, parser)
       SQLFunction(
         name = function.identifier,
-        inputParam = props.get(INPUT_PARAM).map(parseTableSchema(_, parser)),
+        inputParam = props.get(INPUT_PARAM).map(parseRoutineParam(_, parser)),
         returnType = returnType.get,
         exprText = props.get(EXPRESSION),
         queryText = props.get(QUERY),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/UserDefinedFunction.scala

@@ -86,6 +86,11 @@ object UserDefinedFunction {
   // The default Hive Metastore SQL schema length for function resource uri.
   private val HIVE_FUNCTION_RESOURCE_URI_LENGTH_THRESHOLD: Int = 4000
 
+  def parseRoutineParam(text: String, parser: ParserInterface): StructType = {
+    val parsed = parser.parseRoutineParam(text)
+    CharVarcharUtils.failIfHasCharVarchar(parsed).asInstanceOf[StructType]
+  }
+
   def parseTableSchema(text: String, parser: ParserInterface): StructType = {
     val parsed = parser.parseTableSchema(text)
     CharVarcharUtils.failIfHasCharVarchar(parsed).asInstanceOf[StructType]

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AbstractSqlParser.scala

@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{CompoundPlanStatement, Logic
 import org.apache.spark.sql.catalyst.trees.Origin
 import org.apache.spark.sql.errors.QueryParsingErrors
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StructType
 
 /**
  * Base class for all ANTLR4 [[ParserInterface]] implementations.
@@ -102,6 +103,13 @@ abstract class AbstractSqlParser extends AbstractParser with ParserInterface {
     }
   }
 
+  override def parseRoutineParam(sqlText: String): StructType = parse(sqlText) { parser =>
+    val ctx = parser.singleRoutineParamList()
+    withErrorHandling(ctx, Some(sqlText)) {
+      astBuilder.visitSingleRoutineParamList(ctx)
+    }
+  }
+
   def withErrorHandling[T](ctx: ParserRuleContext, sqlText: Option[String])(toResult: => T): T = {
     withOrigin(ctx, sqlText) {
       try {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -25,7 +25,6 @@ import scala.jdk.CollectionConverters._
 import scala.util.{Left, Right}
 
 import org.antlr.v4.runtime.{ParserRuleContext, RuleContext, Token}
-import org.antlr.v4.runtime.misc.Interval
 import org.antlr.v4.runtime.tree.{ParseTree, RuleNode, TerminalNode}
 
 import org.apache.spark.{SparkArithmeticException, SparkException, SparkIllegalArgumentException, SparkThrowable, SparkThrowableHelper}
@@ -117,6 +116,21 @@ class AstBuilder extends DataTypeAstBuilder
     }
   }
 
+  /**
+   * Retrieves the original input text for a given parser context, preserving all whitespace and
+   * formatting.
+   *
+   * ANTLR's default getText method removes whitespace because lexer rules typically skip it.
+   * This utility method extracts the exact text from the original input stream, using token
+   * indices.
+   *
+   * @param ctx The parser context to retrieve original text from.
+   * @return The original input text, including all whitespaces and formatting.
+   */
+  private def getOriginalText(ctx: ParserRuleContext): String = {
+    SparkParserUtils.source(ctx)
+  }
+
   /**
    * Override the default behavior for all visit methods. This will only return a non-null result
    * when the context has only one child. This is done because there is no generic method to
@@ -3848,6 +3862,29 @@ class AstBuilder extends DataTypeAstBuilder
    * DataType parsing
    * ******************************************************************************************** */
 
+  override def visitSingleRoutineParamList(
+      ctx: SingleRoutineParamListContext): StructType = withOrigin(ctx) {
+    val cols = visitColDefinitionList(ctx.colDefinitionList())
+    // Generated columns should have been rejected by the parser.
+    for (col <- cols) {
+      assert(col.generationExpression.isEmpty)
+      assert(col.identityColumnSpec.isEmpty)
+    }
+    // Build fields from the columns, converting comments and default values
+    val fields = for (col <- cols) yield {
+      val metadataBuilder = new MetadataBuilder().withMetadata(col.metadata)
+      col.comment.foreach { c =>
+        metadataBuilder.putString("comment", c)
+      }
+      col.defaultValue.foreach { default =>
+        metadataBuilder.putString(
+          StructType.SQL_FUNCTION_DEFAULT_METADATA_KEY, default.originalSQL)
+      }
+      StructField(col.name, col.dataType, col.nullable, metadataBuilder.build())
+    }
+    StructType(fields.toArray)
+  }
+
   /**
    * Create top level table schema.
    */
@@ -3950,14 +3987,7 @@ class AstBuilder extends DataTypeAstBuilder
     if (expr.containsPattern(PARAMETER)) {
       throw QueryParsingErrors.parameterMarkerNotAllowed(place, expr.origin)
     }
-    // Extract the raw expression text so that we can save the user provided text. We don't
-    // use `Expression.sql` to avoid storing incorrect text caused by bugs in any expression's
-    // `sql` method. Note: `exprCtx.getText` returns a string without spaces, so we need to
-    // get the text from the underlying char stream instead.
-    val start = exprCtx.getStart.getStartIndex
-    val end = exprCtx.getStop.getStopIndex
-    val originalSQL = exprCtx.getStart.getInputStream.getText(new Interval(start, end))
-    DefaultValueExpression(expr, originalSQL)
+    DefaultValueExpression(expr, getOriginalText(exprCtx))
   }
 
   /**