[SPARK-38692][SQL] Use error classes in the compilation errors of function args

core/src/main/resources/error/error-classes.json

@@ -105,6 +105,10 @@
     "message" : [ "The fraction of sec must be zero. Valid range is [0, 60]. If necessary set <config> to false to bypass this error. " ],
     "sqlState" : "22023"
   },
+  "INVALID_FUNCTION_ARGUMENTS" : {
+    "message" : [ "Invalid <invalidType> of arguments for function <functionName>. <message>" ],
+    "sqlState" : "22023"
+  },
   "INVALID_JSON_SCHEMA_MAPTYPE" : {
     "message" : [ "Input schema <dataType> can only contain StringType as a key type for a MapType." ]
   },

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -153,6 +153,9 @@ object FunctionRegistryBase {
         } catch {
           // the exception is an invocation exception. To get a meaningful message, we need the
           // cause.
+          case e: Exception if e.getCause.isInstanceOf[AnalysisException] =>
+            if (e.getCause.asInstanceOf[AnalysisException].errorClass.isDefined) throw e.getCause
+            else throw new AnalysisException(e.getCause.getMessage)
           case e: Exception => throw new AnalysisException(e.getCause.getMessage)
         }
       }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxCountDistinctForIntervals.scala

@@ -55,7 +55,8 @@ case class ApproxCountDistinctForIntervals(
     this(
       child = child,
       endpointsExpression = endpointsExpression,
-      relativeSD = HyperLogLogPlusPlus.validateDoubleLiteral(relativeSD),
+      relativeSD = HyperLogLogPlusPlus
+        .validateDoubleLiteral("ApproxCountDistinctForIntervals", relativeSD),
       mutableAggBufferOffset = 0,
       inputAggBufferOffset = 0)
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/HyperLogLogPlusPlus.scala

@@ -70,7 +70,7 @@ case class HyperLogLogPlusPlus(
   def this(child: Expression, relativeSD: Expression) = {
     this(
       child = child,
-      relativeSD = HyperLogLogPlusPlus.validateDoubleLiteral(relativeSD),
+      relativeSD = HyperLogLogPlusPlus.validateDoubleLiteral("approx_count_distinct", relativeSD),
       mutableAggBufferOffset = 0,
       inputAggBufferOffset = 0)
   }
@@ -144,10 +144,10 @@ case class HyperLogLogPlusPlus(
 }
 
 object HyperLogLogPlusPlus {
-  def validateDoubleLiteral(exp: Expression): Double = exp match {
+  def validateDoubleLiteral(name: String, exp: Expression): Double = exp match {
     case Literal(d: Double, DoubleType) => d
     case Literal(dec: Decimal, _) => dec.toDouble
     case _ =>
-      throw QueryCompilationErrors.secondArgumentNotDoubleLiteralError
+      throw QueryCompilationErrors.secondArgumentNotDoubleLiteralError(name)
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -442,14 +442,17 @@ object QueryCompilationErrors extends QueryErrorsBase {
 
   def invalidFunctionArgumentsError(
       name: String, expectedInfo: String, actualNumber: Int): Throwable = {
-    new AnalysisException(s"Invalid number of arguments for function $name. " +
-      s"Expected: $expectedInfo; Found: $actualNumber")
+    new AnalysisException(
+      errorClass = "INVALID_FUNCTION_ARGUMENTS",
+      messageParameters = Array("number", name, s"Expected: $expectedInfo; Found: $actualNumber"))
   }
 
   def invalidFunctionArgumentNumberError(
       validParametersCount: Seq[Int], name: String, actualNumber: Int): Throwable = {
     if (validParametersCount.length == 0) {
-      new AnalysisException(s"Invalid arguments for function $name")
+      new AnalysisException(
+        errorClass = "INVALID_FUNCTION_ARGUMENTS",
+        messageParameters = Array("number", name, ""))
     } else {
       val expectedNumberOfParameters = if (validParametersCount.length == 1) {
         validParametersCount.head.toString
@@ -462,7 +465,9 @@ object QueryCompilationErrors extends QueryErrorsBase {
   }
 
   def functionAcceptsOnlyOneArgumentError(name: String): Throwable = {
-    new AnalysisException(s"Function $name accepts only one argument")
+    new AnalysisException(
+      errorClass = "INVALID_FUNCTION_ARGUMENTS",
+      messageParameters = Array("number", name, s"It accepts only one argument"))
   }
 
   def alterV2TableSetLocationWithPartitionNotSupportedError(): Throwable = {
@@ -765,8 +770,12 @@ object QueryCompilationErrors extends QueryErrorsBase {
     new AnalysisException(s"Unsupported component type $clz in arrays")
   }
 
-  def secondArgumentNotDoubleLiteralError(): Throwable = {
-    new AnalysisException("The second argument should be a double literal.")
+  def secondArgumentNotDoubleLiteralError(name: String): Throwable = {
+    new AnalysisException(
+      errorClass = "INVALID_FUNCTION_ARGUMENTS",
+      messageParameters = Array("type", name,
+        "The second argument should be a double literal")
+    )
   }
 
   def dataTypeUnsupportedByExtractValueError(
@@ -1479,16 +1488,24 @@ object QueryCompilationErrors extends QueryErrorsBase {
       unbound: UnboundFunction,
       arguments: Seq[Expression],
       unsupported: UnsupportedOperationException): Throwable = {
-    new AnalysisException(s"Function '${unbound.name}' cannot process " +
-      s"input: (${arguments.map(_.dataType.simpleString).mkString(", ")}): " +
-      unsupported.getMessage, cause = Some(unsupported))
+    new AnalysisException(
+      errorClass = "INVALID_FUNCTION_ARGUMENTS",
+      messageParameters = Array("operation", unbound.name,
+        s"It cannot process " +
+        s"input: (${arguments.map(x => toSQLType(x.dataType)).mkString(", ")}): " +
+          unsupported.getMessage),
+      cause = Some(unsupported))
   }
 
   def v2FunctionInvalidInputTypeLengthError(
       bound: BoundFunction,
       args: Seq[Expression]): Throwable = {
-    new AnalysisException(s"Invalid bound function '${bound.name()}: there are ${args.length} " +
-        s"arguments but ${bound.inputTypes().length} parameters returned from 'inputTypes()'")
+    new AnalysisException(
+      errorClass = "INVALID_FUNCTION_ARGUMENTS",
+      messageParameters = Array("number", bound.name(),
+        s"There are ${args.length} arguments " +
+          s"but ${bound.inputTypes().length} parameters returned from 'inputTypes()'")
+    )
   }
 
   def ambiguousRelationAliasNameInNestedCTEError(name: String): Throwable = {
@@ -1568,7 +1585,11 @@ object QueryCompilationErrors extends QueryErrorsBase {
   }
 
   def secondArgumentInFunctionIsNotBooleanLiteralError(funcName: String): Throwable = {
-    new AnalysisException(s"The second argument in $funcName should be a boolean literal.")
+    new AnalysisException(
+      errorClass = "INVALID_FUNCTION_ARGUMENTS",
+      messageParameters = Array("type", funcName,
+        "The second argument should be a boolean literal")
+    )
   }
 
   def joinConditionMissingOrTrivialError(

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/FirstLastTestSuite.scala

@@ -112,10 +112,10 @@ class FirstLastTestSuite extends SparkFunSuite {
     val msg1 = intercept[AnalysisException] {
       new First(input, Literal(1, IntegerType))
     }.getMessage
-    assert(msg1.contains("The second argument in first should be a boolean literal"))
+    assert(msg1.contains("The second argument should be a boolean literal"))
     val msg2 = intercept[AnalysisException] {
       new Last(input, Literal(1, IntegerType))
     }.getMessage
-    assert(msg2.contains("The second argument in last should be a boolean literal"))
+    assert(msg2.contains("The second argument should be a boolean literal"))
   }
 }

sql/core/src/test/resources/sql-tests/results/ansi/string-functions.sql.out

@@ -667,7 +667,7 @@ select decode()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function decode. Expected: 2; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function decode. Expected: 2; Found: 0; line 1 pos 7
 
 
 -- !query
@@ -676,7 +676,7 @@ select decode(encode('abc', 'utf-8'))
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function decode. Expected: 2; Found: 1; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function decode. Expected: 2; Found: 1; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/ceil-floor-with-scale-param.sql.out

@@ -98,7 +98,7 @@ SELECT CEIL(2.5, 0, 0)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function ceil. Expected: 2; Found: 3; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function ceil. Expected: 2; Found: 3; line 1 pos 7
 
 
 -- !query
@@ -197,4 +197,4 @@ SELECT FLOOR(2.5, 0, 0)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function floor. Expected: 2; Found: 3; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function floor. Expected: 2; Found: 3; line 1 pos 7

sql/core/src/test/resources/sql-tests/results/csv-functions.sql.out

@@ -73,7 +73,7 @@ select from_csv()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function from_csv. Expected: one of 2 and 3; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function from_csv. Expected: one of 2 and 3; Found: 0; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/json-functions.sql.out

@@ -90,7 +90,7 @@ select to_json()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function to_json. Expected: one of 1 and 2; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function to_json. Expected: one of 1 and 2; Found: 0; line 1 pos 7
 
 
 -- !query
@@ -164,7 +164,7 @@ select from_json()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function from_json. Expected: one of 2 and 3; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function from_json. Expected: one of 2 and 3; Found: 0; line 1 pos 7
 
 
 -- !query
@@ -435,7 +435,7 @@ select json_array_length()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function json_array_length. Expected: 1; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function json_array_length. Expected: 1; Found: 0; line 1 pos 7
 
 
 -- !query
@@ -508,7 +508,7 @@ select json_object_keys()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function json_object_keys. Expected: 1; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function json_object_keys. Expected: 1; Found: 0; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/sql-compatibility-functions.sql.out

@@ -96,7 +96,7 @@ SELECT string(1, 2)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Function string accepts only one argument; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function string. It accepts only one argument; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/string-functions.sql.out

@@ -651,7 +651,7 @@ select decode()
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function decode. Expected: 2; Found: 0; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function decode. Expected: 2; Found: 0; line 1 pos 7
 
 
 -- !query
@@ -660,7 +660,7 @@ select decode(encode('abc', 'utf-8'))
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function decode. Expected: 2; Found: 1; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function decode. Expected: 2; Found: 1; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/table-valued-functions.sql.out

@@ -75,7 +75,7 @@ Table-valued function range with alternatives:
     range(start: long, end: long, step: long)
     range(start: long, end: long)
     range(end: long)
-cannot be applied to (integer, integer, integer, integer, integer): Invalid number of arguments for function range. Expected: one of 1, 2, 3 and 4; Found: 5; line 1 pos 14
+cannot be applied to (integer, integer, integer, integer, integer): [INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function range. Expected: one of 1, 2, 3 and 4; Found: 5; line 1 pos 14
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/timestamp-ntz.sql.out

@@ -48,7 +48,7 @@ SELECT make_timestamp_ntz(2021, 07, 11, 6, 30, 45.678, 'CET')
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function make_timestamp_ntz. Expected: 6; Found: 7; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function make_timestamp_ntz. Expected: 6; Found: 7; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/udaf.sql.out

@@ -34,7 +34,7 @@ SELECT default.myDoubleAvg(int_col1, 3) as my_avg from t1
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function default.myDoubleAvg. Expected: 1; Found: 2; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function default.myDoubleAvg. Expected: 1; Found: 2; line 1 pos 7
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/udf/udf-udaf.sql.out

@@ -34,7 +34,7 @@ SELECT default.myDoubleAvg(udf(int_col1), udf(3)) as my_avg from t1
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-Invalid number of arguments for function default.myDoubleAvg. Expected: 1; Found: 2; line 1 pos 7
+[INVALID_FUNCTION_ARGUMENTS] Invalid number of arguments for function default.myDoubleAvg. Expected: 1; Found: 2; line 1 pos 7
 
 
 -- !query

sql/core/src/test/scala/org/apache/spark/sql/UDFSuite.scala

@@ -618,7 +618,7 @@ class UDFSuite extends QueryTest with SharedSparkSession {
     val e = intercept[AnalysisException] {
       spark.sql("SELECT CAST(1)")
     }
-    assert(e.getMessage.contains("Invalid arguments for function cast"))
+    assert(e.getMessage.contains("Invalid number of arguments for function cast"))
   }
 
   test("only one case class parameter") {