[SPARK-40358][SQL] Migrate collection type check failures onto error classes

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

@@ -86,6 +86,11 @@
       "Cannot resolve <sqlExpr> due to data type mismatch:"
     ],
     "subClass" : {
+      "BINARY_ARRAY_DIFF_TYPES" : {
+        "message" : [
+          "Input to function <functionName> should have been two <arrayType> with same element type, but it's [<leftType>, <rightType>]."
+        ]
+      },
       "BINARY_OP_DIFF_TYPES" : {
         "message" : [
           "the left and right operands of the binary operator have incompatible types (<left> and <right>)."
@@ -118,6 +123,11 @@
           "To convert values from <srcType> to <targetType>, you can use the functions <functionNames> instead."
         ]
       },
+      "DATA_DIFF_TYPES" : {
+        "message" : [
+          "Input to <functionName> should all be the same type, but it's <dataType>."
+        ]
+      },
       "FRAME_LESS_OFFSET_WITHOUT_FOLDABLE" : {
         "message" : [
           "Offset expression <offset> must be a literal."
@@ -133,6 +143,31 @@
           "Input schema <schema> must be a struct, an array or a map."
         ]
       },
+      "INVALID_MAP_KEY_TYPE" : {
+        "message" : [
+          "The key of map cannot be/contain <keyType>."
+        ]
+      },
+      "INVALID_ORDERING_TYPE" : {
+        "message" : [
+          "The <functionName> does not support ordering on type <dataType>."
+        ]
+      },
+      "MAP_CONCAT_DIFF_TYPES" : {
+        "message" : [
+          "The <functionName> should all be of type map, but it's <dataType>."
+        ]
+      },
+      "MAP_CONTAINS_KEY_DIFF_TYPES" : {
+        "message" : [
+          "Input to <functionName> should have been <dataType> followed by a value with same key type, but it's [<leftType>, <rightType>]."
+        ]
+      },
+      "MAP_FROM_ENTRIES_WRONG_TYPE" : {
+        "message" : [
+          "The <functionName> accepts only arrays of pair structs, but <childExpr> is of <childType>."
+        ]
+      },
       "NON_FOLDABLE_INPUT" : {
         "message" : [
           "the input should be a foldable string expression and not null; however, got <inputExpr>."
@@ -143,6 +178,11 @@
           "all arguments must be strings."
         ]
       },
+      "NULL_TYPE" : {
+        "message" : [
+          "Null typed values cannot be used as arguments of <functionName>."
+        ]
+      },
       "RANGE_FRAME_INVALID_TYPE" : {
         "message" : [
           "The data type <orderSpecType> used in the order specification does not match the data type <valueBoundaryType> which is used in the range frame."

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/collectionOperations.scala

@@ -24,7 +24,9 @@ import scala.reflect.ClassTag
 
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{TypeCheckResult, TypeCoercion, UnresolvedAttribute, UnresolvedSeed}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
 import org.apache.spark.sql.catalyst.expressions.ArraySortLike.NullOrder
+import org.apache.spark.sql.catalyst.expressions.Cast.{toSQLExpr, toSQLType}
 import org.apache.spark.sql.catalyst.expressions.codegen._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.catalyst.trees.{BinaryLike, SQLQueryContext, UnaryLike}
@@ -66,9 +68,16 @@ trait BinaryArrayExpressionWithImplicitCast extends BinaryExpression
     (left.dataType, right.dataType) match {
       case (ArrayType(e1, _), ArrayType(e2, _)) if e1.sameType(e2) =>
         TypeCheckResult.TypeCheckSuccess
-      case _ => TypeCheckResult.TypeCheckFailure(s"input to function $prettyName should have " +
-        s"been two ${ArrayType.simpleString}s with same element type, but it's " +
-        s"[${left.dataType.catalogString}, ${right.dataType.catalogString}]")
+      case _ =>
+        DataTypeMismatch(
+          errorSubClass = "BINARY_ARRAY_DIFF_TYPES",
+          messageParameters = Map(
+            "functionName" -> prettyName,
+            "arrayType" -> toSQLType(ArrayType),
+            "leftType" -> toSQLType(left.dataType),
+            "rightType" -> toSQLType(right.dataType)
+          )
+        )
     }
   }
 
@@ -229,12 +238,21 @@ case class MapContainsKey(left: Expression, right: Expression)
   override def checkInputDataTypes(): TypeCheckResult = {
     (left.dataType, right.dataType) match {
       case (_, NullType) =>
-        TypeCheckResult.TypeCheckFailure("Null typed values cannot be used as arguments")
+        DataTypeMismatch(
+          errorSubClass = "NULL_TYPE",
+          Map("functionName" -> prettyName))
       case (MapType(kt, _, _), dt) if kt.sameType(dt) =>
         TypeUtils.checkForOrderingExpr(kt, s"function $prettyName")
-      case _ => TypeCheckResult.TypeCheckFailure(s"Input to function $prettyName should have " +
-        s"been ${MapType.simpleString} followed by a value with same key type, but it's " +
-        s"[${left.dataType.catalogString}, ${right.dataType.catalogString}].")
+      case _ =>
+        DataTypeMismatch(
+          errorSubClass = "MAP_CONTAINS_KEY_DIFF_TYPES",
+          messageParameters = Map(
+            "functionName" -> prettyName,
+            "dataType" -> toSQLType(MapType),
+            "leftType" -> toSQLType(left.dataType),
+            "rightType" -> toSQLType(right.dataType)
+          )
+        )
     }
   }
 
@@ -663,9 +681,13 @@ case class MapConcat(children: Seq[Expression]) extends ComplexTypeMergingExpres
   override def checkInputDataTypes(): TypeCheckResult = {
     val funcName = s"function $prettyName"
     if (children.exists(!_.dataType.isInstanceOf[MapType])) {
-      TypeCheckResult.TypeCheckFailure(
-        s"input to $funcName should all be of type map, but it's " +
-          children.map(_.dataType.catalogString).mkString("[", ", ", "]"))
+      DataTypeMismatch(
+        errorSubClass = "MAP_CONCAT_DIFF_TYPES",
+        messageParameters = Map(
+          "functionName" -> funcName,
+          "dataType" -> children.map(_.dataType).map(toSQLType).mkString("[", ", ", "]")
+        )
+      )
     } else {
       val sameTypeCheck = TypeUtils.checkForSameTypeInputExpr(children.map(_.dataType), funcName)
       if (sameTypeCheck.isFailure) {
@@ -801,8 +823,15 @@ case class MapFromEntries(child: Expression) extends UnaryExpression with NullIn
   override def checkInputDataTypes(): TypeCheckResult = dataTypeDetails match {
     case Some((mapType, _, _)) =>
       TypeUtils.checkForMapKeyType(mapType.keyType)
-    case None => TypeCheckResult.TypeCheckFailure(s"'${child.sql}' is of " +
-      s"${child.dataType.catalogString} type. $prettyName accepts only arrays of pair structs.")
+    case None =>
+      DataTypeMismatch(
+        errorSubClass = "MAP_FROM_ENTRIES_WRONG_TYPE",
+        messageParameters = Map(
+          "functionName" -> prettyName,
+          "childExpr" -> toSQLExpr(child),
+          "childType" -> toSQLType(child.dataType)
+        )
+      )
   }
 
   private lazy val mapBuilder = new ArrayBasedMapBuilder(dataType.keyType, dataType.valueType)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/TypeUtils.scala

@@ -18,6 +18,8 @@
 package org.apache.spark.sql.catalyst.util
 
 import org.apache.spark.sql.catalyst.analysis.{TypeCheckResult, TypeCoercion}
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
+import org.apache.spark.sql.catalyst.expressions.Cast.toSQLType
 import org.apache.spark.sql.catalyst.expressions.RowOrdering
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.types._
@@ -31,24 +33,38 @@ object TypeUtils {
     if (RowOrdering.isOrderable(dt)) {
       TypeCheckResult.TypeCheckSuccess
     } else {
-      TypeCheckResult.TypeCheckFailure(
-        s"$caller does not support ordering on type ${dt.catalogString}")
+      DataTypeMismatch(
+        errorSubClass = "INVALID_ORDERING_TYPE",
+        Map(
+          "functionName" -> caller,
+          "dataType" -> toSQLType(dt)
+        )
+      )
     }
   }
 
   def checkForSameTypeInputExpr(types: Seq[DataType], caller: String): TypeCheckResult = {
     if (TypeCoercion.haveSameType(types)) {
       TypeCheckResult.TypeCheckSuccess
     } else {
-      TypeCheckResult.TypeCheckFailure(
-        s"input to $caller should all be the same type, but it's " +
-          types.map(_.catalogString).mkString("[", ", ", "]"))
+      DataTypeMismatch(
+        errorSubClass = "DATA_DIFF_TYPES",
+        messageParameters = Map(
+          "functionName" -> caller,
+          "dataType" -> types.map(toSQLType).mkString("(", " or ", ")")
+        )
+      )
     }
   }
 
   def checkForMapKeyType(keyType: DataType): TypeCheckResult = {
     if (keyType.existsRecursively(_.isInstanceOf[MapType])) {
-      TypeCheckResult.TypeCheckFailure("The key of map cannot be/contain map.")
+      DataTypeMismatch(
+        errorSubClass = "INVALID_MAP_KEY_TYPE",
+        messageParameters = Map(
+          "keyType" -> toSQLType(keyType)
+        )
+      )
     } else {
       TypeCheckResult.TypeCheckSuccess
     }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala

@@ -719,7 +719,17 @@ class AnalysisErrorSuite extends AnalysisTest {
       right,
       joinType = Cross,
       condition = Some($"b" === $"d"))
-    assertAnalysisError(plan2, "EqualTo does not support ordering on type map" :: Nil)
+
+    assertAnalysisErrorClass(
+      inputPlan = plan2,
+      expectedErrorClass = "DATATYPE_MISMATCH.INVALID_ORDERING_TYPE",
+      expectedMessageParameters = Map(
+        "functionName" -> "EqualTo",
+        "dataType" -> "\"MAP<STRING, STRING>\"",
+        "sqlExpr" -> "\"(b = d)\""
+      ),
+      caseSensitive = true
+    )
   }
 
   test("PredicateSubQuery is used outside of a allowed nodes") {