[SPARK-37941][SQL] Use error classes in the compilation errors of casting

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

@@ -15,6 +15,9 @@
     "message" : [ "Cannot parse decimal" ],
     "sqlState" : "42000"
   },
+  "CANNOT_UP_CAST_DATATYPE" : {
+    "message" : [ "Cannot up cast %s from %s to %s.\n%s" ]
+  },
   "CAST_CAUSES_OVERFLOW" : {
     "message" : [ "Casting %s to %s causes overflow. To return NULL instead, use 'try_cast'. If necessary set %s to false to bypass this error." ],
     "sqlState" : "22005"

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

@@ -158,16 +158,24 @@ object QueryCompilationErrors {
   def upCastFailureError(
       fromStr: String, from: Expression, to: DataType, walkedTypePath: Seq[String]): Throwable = {
     new AnalysisException(
-      s"Cannot up cast $fromStr from " +
-        s"${from.dataType.catalogString} to ${to.catalogString}.\n" +
+      errorClass = "CANNOT_UP_CAST_DATATYPE",
+      messageParameters = Array(
+        fromStr,
+        from.dataType.catalogString,
+        to.catalogString,
         s"The type path of the target object is:\n" + walkedTypePath.mkString("", "\n", "\n") +
-        "You can either add an explicit cast to the input data or choose a higher precision " +
-        "type of the field in the target object")
+          "You can either add an explicit cast to the input data or choose a higher precision " +
+          "type of the field in the target object"
+      )
+    )
   }
 
   def unsupportedAbstractDataTypeForUpCastError(gotType: AbstractDataType): Throwable = {
     new AnalysisException(
-      s"UpCast only support DecimalType as AbstractDataType yet, but got: $gotType")
+      errorClass = "UNSUPPORTED_FEATURE",
+      messageParameters =
+        Array(s"UpCast only support DecimalType as AbstractDataType yet, but got: $gotType")
+    )
   }
 
   def outerScopeFailureForNewInstanceError(className: String): Throwable = {
@@ -415,13 +423,6 @@ object QueryCompilationErrors {
         s"'${child.output.map(_.name).mkString("(", ",", ")")}'")
   }
 
-  def cannotUpCastAsAttributeError(
-      fromAttr: Attribute, toAttr: Attribute): Throwable = {
-    new AnalysisException(s"Cannot up cast ${fromAttr.sql} from " +
-      s"${fromAttr.dataType.catalogString} to ${toAttr.dataType.catalogString} " +
-      "as it may truncate")
-  }
-
   def functionUndefinedError(name: FunctionIdentifier): Throwable = {
     new AnalysisException(s"undefined function $name")
   }

sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala

@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.errors
+
+import org.apache.spark.sql.{AnalysisException, Dataset, QueryTest}
+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
+import org.apache.spark.sql.catalyst.expressions.{Alias, UpCast}
+import org.apache.spark.sql.catalyst.plans.logical.Project
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.NumericType
+
+case class StringLongClass(a: String, b: Long)
+
+case class StringIntClass(a: String, b: Int)
+
+case class ComplexClass(a: Long, b: StringLongClass)
+
+class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
+  import testImplicits._
+
+  test("CANNOT_UP_CAST_DATATYPE: invalid upcast data type") {
+    val msg1 = intercept[AnalysisException] {
+      sql("select 'value1' as a, 1L as b").as[StringIntClass]
+    }.message
+    assert(msg1 ==
+      s"""
+         |Cannot up cast b from bigint to int.
+         |The type path of the target object is:
+         |- field (class: "scala.Int", name: "b")
+         |- root class: "org.apache.spark.sql.errors.StringIntClass"
+         |You can either add an explicit cast to the input data or choose a higher precision type
+       """.stripMargin.trim + " of the field in the target object")
+
+    val msg2 = intercept[AnalysisException] {
+      sql("select 1L as a," +
+        " named_struct('a', 'value1', 'b', cast(1.0 as decimal(38,18))) as b")
+        .as[ComplexClass]
+    }.message
+    assert(msg2 ==
+      s"""
+         |Cannot up cast b.`b` from decimal(38,18) to bigint.
+         |The type path of the target object is:
+         |- field (class: "scala.Long", name: "b")
+         |- field (class: "org.apache.spark.sql.errors.StringLongClass", name: "b")
+         |- root class: "org.apache.spark.sql.errors.ComplexClass"
+         |You can either add an explicit cast to the input data or choose a higher precision type
+       """.stripMargin.trim + " of the field in the target object")
+  }
+
+  test("UNSUPPORTED_FEATURE: UpCast only support DecimalType as AbstractDataType") {
+    val df = sql("select 1 as value")
+
+    val msg = intercept[AnalysisException] {
+      val plan = Project(
+        Seq(Alias(UpCast(UnresolvedAttribute("value"), NumericType), "value")()),
+        df.logicalPlan)
+
+      Dataset.ofRows(spark, plan)
+    }.message
+    assert(msg.contains("The feature is not supported: " +
+      "UpCast only support DecimalType as AbstractDataType yet," +
+      " but got: org.apache.spark.sql.types.NumericType"))
+  }
+
+}