diff --git a/core/src/main/java/org/apache/spark/SparkThrowable.java b/core/src/main/java/org/apache/spark/SparkThrowable.java
index 2be0c3c0f9466..581e1f6eebb8b 100644
--- a/core/src/main/java/org/apache/spark/SparkThrowable.java
+++ b/core/src/main/java/org/apache/spark/SparkThrowable.java
@@ -36,6 +36,10 @@ public interface SparkThrowable {
   // If null, error class is not set
   String getErrorClass();
 
+  default String getErrorSubClass() {
+    return null;
+  }
+
   // Portable error identifier across SQL engines
   // If null, error class or SQLSTATE is not set
   default String getSqlState() {
@@ -46,4 +50,13 @@ default String getSqlState() {
   default boolean isInternalError() {
     return SparkThrowableHelper.isInternalError(this.getErrorClass());
   }
+
+  default String[] getMessageParameters() {
+    return new String[]{};
+  }
+
+  // Returns a string array of all parameters that need to be passed to this error message.
+  default String[] getParameterNames() {
+    return SparkThrowableHelper.getParameterNames(this.getErrorClass(), this.getErrorSubClass());
+  }
 }
diff --git a/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java b/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java
index c5f19a0c2012b..9d2739018a038 100644
--- a/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java
+++ b/core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java
@@ -39,11 +39,17 @@ public SparkOutOfMemoryError(OutOfMemoryError e) {
     }
 
     public SparkOutOfMemoryError(String errorClass, String[] messageParameters) {
-        super(SparkThrowableHelper.getMessage(errorClass, messageParameters, ""));
+        super(SparkThrowableHelper.getMessage(errorClass, null,
+                messageParameters, ""));
         this.errorClass = errorClass;
         this.messageParameters = messageParameters;
     }
 
+    @Override
+    public String[] getMessageParameters() {
+        return messageParameters;
+    }
+
     @Override
     public String getErrorClass() {
         return errorClass;
diff --git a/core/src/main/resources/error/error-classes.json b/core/src/main/resources/error/error-classes.json
index 59553e78d74ab..833ecc0a3c09e 100644
--- a/core/src/main/resources/error/error-classes.json
+++ b/core/src/main/resources/error/error-classes.json
@@ -31,19 +31,19 @@
   },
   "CANNOT_UP_CAST_DATATYPE" : {
     "message" : [
-      "Cannot up cast <value> from <sourceType> to <targetType>.",
+      "Cannot up cast <expression> from <sourceType> to <targetType>.",
       "<details>"
     ]
   },
   "CAST_INVALID_INPUT" : {
     "message" : [
-      "The value <value> of the type <sourceType> cannot be cast to <targetType> because it is malformed. Correct the value as per the syntax, or change its target type. Use `try_cast` to tolerate malformed input and return NULL instead. If necessary set <config> to \"false\" to bypass this error."
+      "The value <expression> of the type <sourceType> cannot be cast to <targetType> because it is malformed. Correct the value as per the syntax, or change its target type. Use `try_cast` to tolerate malformed input and return NULL instead. If necessary set <ansiConfig> to \"false\" to bypass this error."
     ],
     "sqlState" : "42000"
   },
   "CAST_OVERFLOW" : {
     "message" : [
-      "The value <value> of the type <sourceType> cannot be cast to <targetType> due to an overflow. Use `try_cast` to tolerate overflow and return NULL instead. If necessary set <config> to \"false\" to bypass this error."
+      "The value <value> of the type <sourceType> cannot be cast to <targetType> due to an overflow. Use `try_cast` to tolerate overflow and return NULL instead. If necessary set <ansiConfig> to \"false\" to bypass this error."
     ],
     "sqlState" : "22005"
   },
@@ -83,7 +83,7 @@
   },
   "FORBIDDEN_OPERATION" : {
     "message" : [
-      "The operation <statement> is not allowed on <objectType>: <objectName>"
+      "The operation <statement> is not allowed on the <objectType>: <objectName>"
     ]
   },
   "GRAPHITE_SINK_INVALID_PROTOCOL" : {
@@ -157,8 +157,7 @@
           "See more details in SPARK-31404. You can set the SQL config <config> or",
           "the datasource option <option> to \"LEGACY\" to rebase the datetime values",
           "w.r.t. the calendar difference during reading. To read the datetime values",
-          "as it is, set the SQL config <config> or the datasource option <option>",
-          "to \"CORRECTED\"."
+          "as it is, set the SQL config or the datasource option to \"CORRECTED\"."
         ]
       },
       "WRITE_ANCIENT_DATETIME" : {
@@ -170,7 +169,7 @@
           "is different from Spark 3.0+'s Proleptic Gregorian calendar. See more",
           "details in SPARK-31404. You can set <config> to \"LEGACY\" to rebase the",
           "datetime values w.r.t. the calendar difference during writing, to get maximum",
-          "interoperability. Or set <config> to \"CORRECTED\" to write the datetime",
+          "interoperability. Or set the config to \"CORRECTED\" to write the datetime",
           "values as it is, if you are sure that the written files will only be read by",
           "Spark 3.0+ or other systems that use Proleptic Gregorian calendar."
         ]
@@ -190,12 +189,12 @@
   },
   "INVALID_ARRAY_INDEX" : {
     "message" : [
-      "The index <indexValue> is out of bounds. The array has <arraySize> elements. If necessary set <config> to \"false\" to bypass this error."
+      "The index <indexValue> is out of bounds. The array has <arraySize> elements. If necessary set <ansiConfig> to \"false\" to bypass this error."
     ]
   },
   "INVALID_ARRAY_INDEX_IN_ELEMENT_AT" : {
     "message" : [
-      "The index <indexValue> is out of bounds. The array has <arraySize> elements. Use `try_element_at` to tolerate accessing element at invalid index and return NULL instead. If necessary set <config> to \"false\" to bypass this error."
+      "The index <indexValue> is out of bounds. The array has <arraySize> elements. Use `try_element_at` to tolerate accessing element at invalid index and return NULL instead. If necessary set <ansiConfig> to \"false\" to bypass this error."
     ]
   },
   "INVALID_BUCKET_FILE" : {
@@ -211,7 +210,7 @@
   },
   "INVALID_FRACTION_OF_SECOND" : {
     "message" : [
-      "The fraction of sec must be zero. Valid range is [0, 60]. If necessary set <config> to \"false\" to bypass this error."
+      "The fraction of sec must be zero. Valid range is [0, 60]. If necessary set <ansiConfig> to \"false\" to bypass this error."
     ],
     "sqlState" : "22023"
   },
@@ -222,7 +221,7 @@
   },
   "INVALID_PANDAS_UDF_PLACEMENT" : {
     "message" : [
-      "The group aggregate pandas UDF <functionName> cannot be invoked together with as other, non-pandas aggregate functions."
+      "The group aggregate pandas UDF <functionList> cannot be invoked together with as other, non-pandas aggregate functions."
     ]
   },
   "INVALID_PARAMETER_VALUE" : {
@@ -266,7 +265,7 @@
   },
   "MULTI_UDF_INTERFACE_ERROR" : {
     "message" : [
-      "Not allowed to implement multiple UDF interfaces, UDF class <class>"
+      "Not allowed to implement multiple UDF interfaces, UDF class <className>"
     ]
   },
   "MULTI_VALUE_SUBQUERY_ERROR" : {
@@ -293,7 +292,7 @@
   },
   "NO_UDF_INTERFACE_ERROR" : {
     "message" : [
-      "UDF class <class> doesn't implement any UDF interface"
+      "UDF class <className> doesn't implement any UDF interface"
     ]
   },
   "PARSE_CHAR_MISSING_LENGTH" : {
@@ -333,7 +332,7 @@
   },
   "SECOND_FUNCTION_ARGUMENT_NOT_INTEGER" : {
     "message" : [
-      "The second argument of '<functionName>' function needs to be an integer."
+      "The second argument of <functionName> function needs to be an integer."
     ],
     "sqlState" : "22023"
   },
@@ -361,7 +360,7 @@
     "subClass" : {
       "DATA_TYPE_MISMATCH" : {
         "message" : [
-          "need <quantifier> <desiredType> field but got <dataType>."
+          "need a(n) <desiredType> field but got <dataType>."
         ]
       },
       "FIELD_NUMBER_MISMATCH" : {
@@ -473,7 +472,7 @@
       },
       "TOO_MANY_TYPE_ARGUMENTS_FOR_UDF_CLASS" : {
         "message" : [
-          "UDF class with <n> type arguments."
+          "UDF class with <num> type arguments."
         ]
       },
       "TRANSFORM_DISTINCT_ALL" : {
@@ -496,17 +495,17 @@
     "subClass" : {
       "MULTI_GENERATOR" : {
         "message" : [
-          "only one generator allowed per <clause> clause but found <size>: <generators>"
+          "only one generator allowed per <clause> clause but found <num>: <generators>"
         ]
       },
       "NESTED_IN_EXPRESSIONS" : {
         "message" : [
-          "nested in expressions <expressions>"
+          "nested in expressions <expression>"
         ]
       },
       "NOT_GENERATOR" : {
         "message" : [
-          "<name> is expected to be a generator. However, its class is <classCanonicalName>, which is not a generator."
+          "<functionName> is expected to be a generator. However, its class is <classCanonicalName>, which is not a generator."
         ]
       },
       "OUTSIDE_SELECT" : {
diff --git a/core/src/main/scala/org/apache/spark/ErrorInfo.scala b/core/src/main/scala/org/apache/spark/ErrorInfo.scala
index 613c591eec2c7..6c6d51456ecfc 100644
--- a/core/src/main/scala/org/apache/spark/ErrorInfo.scala
+++ b/core/src/main/scala/org/apache/spark/ErrorInfo.scala
@@ -73,6 +73,7 @@ private[spark] object SparkThrowableHelper {
 
   def getMessage(
       errorClass: String,
+      errorSubClass: String,
       messageParameters: Array[String],
       queryContext: String = ""): String = {
     val errorInfo = errorClassToInfoMap.getOrElse(errorClass,
@@ -80,11 +81,13 @@ private[spark] object SparkThrowableHelper {
     val (displayClass, displayMessageParameters, displayFormat) = if (errorInfo.subClass.isEmpty) {
       (errorClass, messageParameters, errorInfo.messageFormat)
     } else {
-      val subClass = errorInfo.subClass.get
-      val subErrorClass = messageParameters.head
-      val errorSubInfo = subClass.getOrElse(subErrorClass,
-        throw new IllegalArgumentException(s"Cannot find sub error class '$subErrorClass'"))
-      (errorClass + "." + subErrorClass, messageParameters.tail,
+      val subClasses = errorInfo.subClass.get
+      if (errorSubClass == null) {
+        throw new IllegalArgumentException(s"Subclass required for error class '$errorClass'")
+      }
+      val errorSubInfo = subClasses.getOrElse(errorSubClass,
+        throw new IllegalArgumentException(s"Cannot find sub error class '$errorSubClass'"))
+      (errorClass + "." + errorSubClass, messageParameters,
         errorInfo.messageFormat + " " + errorSubInfo.messageFormat)
     }
     val displayMessage = String.format(
@@ -98,6 +101,29 @@ private[spark] object SparkThrowableHelper {
     s"[$displayClass] $displayMessage$displayQueryContext"
   }
 
+  def getParameterNames(errorClass: String, errorSubCLass: String): Array[String] = {
+    val errorInfo = errorClassToInfoMap.getOrElse(errorClass,
+      throw new IllegalArgumentException(s"Cannot find error class '$errorClass'"))
+    if (errorInfo.subClass.isEmpty && errorSubCLass != null) {
+      throw new IllegalArgumentException(s"'$errorClass' has no subclass")
+    }
+    if (errorInfo.subClass.isDefined && errorSubCLass == null) {
+      throw new IllegalArgumentException(s"'$errorClass' requires subclass")
+    }
+    var parameterizedMessage = errorInfo.messageFormat
+    if (errorInfo.subClass.isDefined) {
+      val givenSubClass = errorSubCLass
+      val errorSubInfo = errorInfo.subClass.get.getOrElse(givenSubClass,
+        throw new IllegalArgumentException(s"Cannot find sub error class '$givenSubClass'"))
+      parameterizedMessage = parameterizedMessage + errorSubInfo.messageFormat
+    }
+    val pattern = "<[a-zA-Z0-9_-]+>".r
+    val matches = pattern.findAllIn(parameterizedMessage)
+    val parameterSeq = matches.toArray
+    val parameterNames = parameterSeq.map(p => p.stripPrefix("<").stripSuffix(">"))
+    parameterNames
+  }
+
   def getSqlState(errorClass: String): String = {
     Option(errorClass).flatMap(errorClassToInfoMap.get).flatMap(_.sqlState).orNull
   }
diff --git a/core/src/main/scala/org/apache/spark/SparkException.scala b/core/src/main/scala/org/apache/spark/SparkException.scala
index c28624cc7a01a..53942054f1b61 100644
--- a/core/src/main/scala/org/apache/spark/SparkException.scala
+++ b/core/src/main/scala/org/apache/spark/SparkException.scala
@@ -28,23 +28,51 @@ class SparkException(
     message: String,
     cause: Throwable,
     errorClass: Option[String],
+    errorSubClass: Option[String],
     messageParameters: Array[String])
   extends Exception(message, cause) with SparkThrowable {
 
+   def this(
+       message: String,
+       cause: Throwable,
+       errorClass: Option[String],
+       messageParameters: Array[String]) =
+     this(message = message,
+          cause = cause,
+          errorClass = errorClass,
+          errorSubClass = None,
+          messageParameters = messageParameters)
+
   def this(message: String, cause: Throwable) =
-    this(message = message, cause = cause, errorClass = None, messageParameters = Array.empty)
+    this(message = message, cause = cause, errorClass = None, errorSubClass = None,
+      messageParameters = Array.empty)
 
   def this(message: String) =
     this(message = message, cause = null)
 
   def this(errorClass: String, messageParameters: Array[String], cause: Throwable) =
     this(
-      message = SparkThrowableHelper.getMessage(errorClass, messageParameters),
+      message = SparkThrowableHelper.getMessage(errorClass, null, messageParameters),
+      cause = cause,
+      errorClass = Some(errorClass),
+      errorSubClass = None,
+      messageParameters = messageParameters)
+
+  def this(
+      errorClass: String,
+      errorSubClass: String,
+      messageParameters: Array[String],
+      cause: Throwable) =
+    this(
+      message = SparkThrowableHelper.getMessage(errorClass, errorSubClass, messageParameters),
       cause = cause,
       errorClass = Some(errorClass),
+      errorSubClass = Some(errorSubClass),
       messageParameters = messageParameters)
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass.orNull
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -73,38 +101,57 @@ private[spark] case class ExecutorDeadException(message: String)
  */
 private[spark] class SparkUpgradeException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     cause: Throwable)
-  extends RuntimeException(SparkThrowableHelper.getMessage(errorClass, messageParameters), cause)
+  extends RuntimeException(SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
+    messageParameters), cause)
     with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Arithmetic exception thrown from Spark with an error class.
  */
 private[spark] class SparkArithmeticException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     queryContext: String = "")
   extends ArithmeticException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
+      messageParameters, queryContext))
     with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Unsupported operation exception thrown from Spark with an error class.
  */
 private[spark] class SparkUnsupportedOperationException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends UnsupportedOperationException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  def this(
+      errorClass: String,
+      errorSubClass: String,
+      messageParameters: Array[String]) =
+    this(
+      errorClass = errorClass,
+      errorSubClass = Some(errorSubClass),
+      messageParameters = messageParameters)
+
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -112,113 +159,141 @@ private[spark] class SparkUnsupportedOperationException(
  */
 private[spark] class SparkClassNotFoundException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     cause: Throwable = null)
   extends ClassNotFoundException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters), cause) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters), cause)
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Concurrent modification exception thrown from Spark with an error class.
  */
 private[spark] class SparkConcurrentModificationException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     cause: Throwable = null)
   extends ConcurrentModificationException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters), cause) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters), cause)
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Datetime exception thrown from Spark with an error class.
  */
 private[spark] class SparkDateTimeException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     queryContext: String = "")
   extends DateTimeException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
+      messageParameters, queryContext))
     with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Hadoop file already exists exception thrown from Spark with an error class.
  */
 private[spark] class SparkFileAlreadyExistsException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends FileAlreadyExistsException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * File not found exception thrown from Spark with an error class.
  */
 private[spark] class SparkFileNotFoundException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends FileNotFoundException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Number format exception thrown from Spark with an error class.
  */
 private[spark] class SparkNumberFormatException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     queryContext: String)
   extends NumberFormatException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
+      messageParameters, queryContext))
     with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * No such method exception thrown from Spark with an error class.
  */
 private[spark] class SparkNoSuchMethodException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends NoSuchMethodException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Illegal argument exception thrown from Spark with an error class.
  */
 private[spark] class SparkIllegalArgumentException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends IllegalArgumentException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
-}
+  override def getErrorSubClass: String = errorSubClass.orNull}
 
 /**
  * Index out of bounds exception thrown from Spark with an error class.
  */
 private[spark] class SparkIndexOutOfBoundsException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends IndexOutOfBoundsException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -226,23 +301,52 @@ private[spark] class SparkIndexOutOfBoundsException(
  */
 private[spark] class SparkIOException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends IOException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 private[spark] class SparkRuntimeException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     cause: Throwable = null,
     queryContext: String = "")
   extends RuntimeException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters, queryContext), cause)
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
+      messageParameters, queryContext),
+    cause)
     with SparkThrowable {
 
+  def this(errorClass: String,
+           errorSubClass: String,
+           messageParameters: Array[String],
+           cause: Throwable,
+           queryContext: String)
+    = this(errorClass = errorClass,
+    errorSubClass = Some(errorSubClass),
+    messageParameters = messageParameters,
+    cause = cause,
+    queryContext = queryContext)
+
+  def this(errorClass: String,
+           errorSubClass: String,
+           messageParameters: Array[String])
+  = this(errorClass = errorClass,
+    errorSubClass = Some(errorSubClass),
+    messageParameters = messageParameters,
+    cause = null,
+    queryContext = "")
+
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -250,11 +354,15 @@ private[spark] class SparkRuntimeException(
  */
 private[spark] class SparkSecurityException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends SecurityException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -262,11 +370,15 @@ private[spark] class SparkSecurityException(
  */
 private[spark] class SparkArrayIndexOutOfBoundsException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends ArrayIndexOutOfBoundsException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -274,11 +386,21 @@ private[spark] class SparkArrayIndexOutOfBoundsException(
  */
 private[spark] class SparkSQLException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends SQLException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
 
+  def this(errorClass: String, messageParameters: Array[String]) =
+    this(
+      errorClass = errorClass,
+      errorSubClass = None,
+      messageParameters = messageParameters)
+
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -286,13 +408,17 @@ private[spark] class SparkSQLException(
  */
 private[spark] class SparkNoSuchElementException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String],
     queryContext: String)
   extends NoSuchElementException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters, queryContext))
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull,
+      messageParameters, queryContext))
     with SparkThrowable {
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
 
 /**
@@ -300,9 +426,20 @@ private[spark] class SparkNoSuchElementException(
  */
 private[spark] class SparkSQLFeatureNotSupportedException(
     errorClass: String,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String])
   extends SQLFeatureNotSupportedException(
-    SparkThrowableHelper.getMessage(errorClass, messageParameters)) with SparkThrowable {
+    SparkThrowableHelper.getMessage(errorClass, errorSubClass.orNull, messageParameters))
+    with SparkThrowable {
+
+  def this(errorClass: String,
+           errorSubClass: String,
+           messageParameters: Array[String]) =
+    this(errorClass = errorClass,
+      errorSubClass = Some(errorSubClass),
+      messageParameters = messageParameters)
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
diff --git a/core/src/test/scala/org/apache/spark/SparkFunSuite.scala b/core/src/test/scala/org/apache/spark/SparkFunSuite.scala
index 02e67c0af1258..7922e13db69bc 100644
--- a/core/src/test/scala/org/apache/spark/SparkFunSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkFunSuite.scala
@@ -264,6 +264,69 @@ abstract class SparkFunSuite
     }
   }
 
+  /**
+   * Checks an exception with an error class against expected results.
+   * @param exception     The exception to check
+   * @param errorClass    The expected error class identifying the error
+   * @param errorSubClass Optional the expected subclass, None if not given
+   * @param sqlState      Optional the expected SQLSTATE, not verified if not supplied
+   * @param parameters    A map of parameter names and values. The names are as defined
+   *                      in the error-classes file.
+   * @param matchPVals    Optionally treat the parameters value as regular expression pattern.
+   *                      false if not supplied.
+   */
+  protected def checkError(
+      exception: SparkThrowable,
+      errorClass: String,
+      errorSubClass: Option[String],
+      sqlState: Option[String],
+      parameters: Map[String, String],
+      matchPVals: Boolean = false): Unit = {
+    assert(exception.getErrorClass === errorClass)
+    if (exception.getErrorSubClass != null) {
+      assert(errorSubClass.isDefined)
+      assert(exception.getErrorSubClass === errorSubClass.get)
+    }
+    sqlState.foreach(state => assert(exception.getSqlState === state))
+    val expectedParameters = (exception.getParameterNames zip exception.getMessageParameters).toMap
+    if (matchPVals == true) {
+      assert(expectedParameters.size === parameters.size)
+      expectedParameters.foreach(
+        exp => {
+          val parm = parameters.getOrElse(exp._1,
+            throw new IllegalArgumentException("Missing parameter" + exp._1))
+          if (!exp._2.matches(parm)) {
+            throw new IllegalArgumentException("(" + exp._1 + ", " + exp._2 +
+              ") does not match: " + parm)
+          }
+        }
+      )
+    } else {
+      assert(expectedParameters === parameters)
+    }
+  }
+
+  protected def checkError(
+      exception: Exception with SparkThrowable,
+      errorClass: String,
+      errorSubClass: String,
+      sqlState: String,
+      parameters: Map[String, String]): Unit =
+    checkError(exception, errorClass, Some(errorSubClass), Some(sqlState), parameters)
+
+  protected def checkError(
+      exception: Exception with SparkThrowable,
+      errorClass: String,
+      sqlState: String,
+      parameters: Map[String, String]): Unit =
+    checkError(exception, errorClass, None, Some(sqlState), parameters)
+
+  protected def checkError(
+      exception: Exception with SparkThrowable,
+      errorClass: String,
+      parameters: Map[String, String]): Unit =
+    checkError(exception, errorClass, None, None, parameters)
+
   class LogAppender(msg: String = "", maxEvents: Int = 1000)
       extends AbstractAppender("logAppender", null, null, true, Property.EMPTY_ARRAY) {
     private val _loggingEvents = new ArrayBuffer[LogEvent]()
diff --git a/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala b/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala
index 85fcd6a9e33a2..91019973dea35 100644
--- a/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkThrowableSuite.scala
@@ -147,12 +147,12 @@ class SparkThrowableSuite extends SparkFunSuite {
 
   test("Check if error class is missing") {
     val ex1 = intercept[IllegalArgumentException] {
-      getMessage("", Array.empty)
+      getMessage("", null, Array.empty)
     }
     assert(ex1.getMessage == "Cannot find error class ''")
 
     val ex2 = intercept[IllegalArgumentException] {
-      getMessage("LOREM_IPSUM", Array.empty)
+      getMessage("LOREM_IPSUM", null, Array.empty)
     }
     assert(ex2.getMessage == "Cannot find error class 'LOREM_IPSUM'")
   }
@@ -160,11 +160,11 @@ class SparkThrowableSuite extends SparkFunSuite {
   test("Check if message parameters match message format") {
     // Requires 2 args
     intercept[IllegalFormatException] {
-      getMessage("MISSING_COLUMN", Array.empty)
+      getMessage("MISSING_COLUMN", null, Array.empty)
     }
 
     // Does not fail with too many args (expects 0 args)
-    assert(getMessage("DIVIDE_BY_ZERO", Array("foo", "bar", "baz")) ==
+    assert(getMessage("DIVIDE_BY_ZERO", null, Array("foo", "bar", "baz")) ==
       "[DIVIDE_BY_ZERO] Division by zero. " +
       "Use `try_divide` to tolerate divisor being 0 and return NULL instead. " +
         "If necessary set foo to \"false\" " +
@@ -172,7 +172,7 @@ class SparkThrowableSuite extends SparkFunSuite {
   }
 
   test("Error message is formatted") {
-    assert(getMessage("MISSING_COLUMN", Array("foo", "bar, baz")) ==
+    assert(getMessage("MISSING_COLUMN", null, Array("foo", "bar, baz")) ==
       "[MISSING_COLUMN] Column 'foo' does not exist. Did you mean one of the following? [bar, baz]")
   }
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala
index 397300d5e7334..9ab0b223e1172 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/AnalysisException.scala
@@ -36,13 +36,32 @@ class AnalysisException protected[sql] (
     @transient val plan: Option[LogicalPlan] = None,
     val cause: Option[Throwable] = None,
     val errorClass: Option[String] = None,
+    val errorSubClass: Option[String] = None,
     val messageParameters: Array[String] = Array.empty)
   extends Exception(message, cause.orNull) with SparkThrowable with Serializable {
 
+    // Needed for binary compatibility
+    protected[sql] def this(message: String,
+           line: Option[Int],
+           startPosition: Option[Int],
+           plan: Option[LogicalPlan],
+           cause: Option[Throwable],
+           errorClass: Option[String],
+           messageParameters: Array[String]) =
+    this(message = message,
+         line = line,
+         startPosition = startPosition,
+         plan = plan,
+         cause = cause,
+         errorClass,
+         errorSubClass = None,
+         messageParameters = messageParameters)
+
   def this(errorClass: String, messageParameters: Array[String], cause: Option[Throwable]) =
     this(
-      SparkThrowableHelper.getMessage(errorClass, messageParameters),
+      SparkThrowableHelper.getMessage(errorClass, null, messageParameters),
       errorClass = Some(errorClass),
+      errorSubClass = None,
       messageParameters = messageParameters,
       cause = cause)
 
@@ -54,10 +73,34 @@ class AnalysisException protected[sql] (
       messageParameters: Array[String],
       origin: Origin) =
     this(
-      SparkThrowableHelper.getMessage(errorClass, messageParameters),
+      SparkThrowableHelper.getMessage(errorClass, null, messageParameters),
+      line = origin.line,
+      startPosition = origin.startPosition,
+      errorClass = Some(errorClass),
+      errorSubClass = None,
+      messageParameters = messageParameters)
+
+  def this(
+      errorClass: String,
+      errorSubClass: String,
+      messageParameters: Array[String]) =
+    this(
+      SparkThrowableHelper.getMessage(errorClass, errorSubClass, messageParameters),
+      errorClass = Some(errorClass),
+      errorSubClass = Some(errorSubClass),
+      messageParameters = messageParameters)
+
+  def this(
+      errorClass: String,
+      errorSubClass: String,
+      messageParameters: Array[String],
+      origin: Origin) =
+    this(
+      SparkThrowableHelper.getMessage(errorClass, errorSubClass, messageParameters),
       line = origin.line,
       startPosition = origin.startPosition,
       errorClass = Some(errorClass),
+      errorSubClass = Some(errorSubClass),
       messageParameters = messageParameters)
 
   def copy(
@@ -68,7 +111,8 @@ class AnalysisException protected[sql] (
       cause: Option[Throwable] = this.cause,
       errorClass: Option[String] = this.errorClass,
       messageParameters: Array[String] = this.messageParameters): AnalysisException =
-    new AnalysisException(message, line, startPosition, plan, cause, errorClass, messageParameters)
+    new AnalysisException(message, line, startPosition, plan, cause, errorClass, errorSubClass,
+      messageParameters)
 
   def withPosition(line: Option[Int], startPosition: Option[Int]): AnalysisException = {
     val newException = this.copy(line = line, startPosition = startPosition)
@@ -91,5 +135,7 @@ class AnalysisException protected[sql] (
     message
   }
 
+  override def getMessageParameters: Array[String] = messageParameters
   override def getErrorClass: String = errorClass.orNull
+  override def getErrorSubClass: String = errorSubClass.orNull
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 087582a924bc4..b8fa6e421ca66 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -2780,7 +2780,7 @@ class Analyzer(override val catalogManager: CatalogManager)
 
       case Project(projectList, _) if projectList.count(hasGenerator) > 1 =>
         val generators = projectList.filter(hasGenerator).map(trimAlias)
-        throw QueryCompilationErrors.moreThanOneGeneratorError(generators, "select")
+        throw QueryCompilationErrors.moreThanOneGeneratorError(generators, "SELECT")
 
       case Aggregate(_, aggList, _) if aggList.exists(hasNestedGenerator) =>
         val nestedGenerator = aggList.find(hasNestedGenerator).get
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InvalidUDFClassException.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InvalidUDFClassException.scala
index 0e5e52a9c9016..28918d1799c3c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InvalidUDFClassException.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/InvalidUDFClassException.scala
@@ -30,5 +30,5 @@ class InvalidUDFClassException private[sql](
   extends AnalysisException(message = message, errorClass = errorClass) {
 
   def this(errorClass: String, messageParameters: Array[String]) =
-    this(SparkThrowableHelper.getMessage(errorClass, messageParameters), Some(errorClass))
+    this(SparkThrowableHelper.getMessage(errorClass, null, messageParameters), Some(errorClass))
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala
index bf0ee9c115d8f..76757f9fc2178 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/ParseDriver.scala
@@ -144,7 +144,7 @@ abstract class AbstractSqlParser extends ParserInterface with SQLConfHelper with
       case e: AnalysisException =>
         val position = Origin(e.line, e.startPosition)
         throw new ParseException(Option(command), e.message, position, position,
-          e.errorClass, e.messageParameters)
+          e.errorClass, None, e.messageParameters)
     }
   }
 }
@@ -237,6 +237,7 @@ class ParseException(
     val start: Origin,
     val stop: Origin,
     errorClass: Option[String] = None,
+    errorSubClass: Option[String] = None,
     messageParameters: Array[String] = Array.empty)
   extends AnalysisException(
     message,
@@ -245,6 +246,7 @@ class ParseException(
     None,
     None,
     errorClass,
+    errorSubClass,
     messageParameters) {
 
   def this(message: String, ctx: ParserRuleContext) = {
@@ -256,10 +258,23 @@ class ParseException(
 
   def this(errorClass: String, messageParameters: Array[String], ctx: ParserRuleContext) =
     this(Option(ParserUtils.command(ctx)),
-      SparkThrowableHelper.getMessage(errorClass, messageParameters),
+      SparkThrowableHelper.getMessage(errorClass, null, messageParameters),
       ParserUtils.position(ctx.getStart),
       ParserUtils.position(ctx.getStop),
       Some(errorClass),
+      None,
+      messageParameters)
+
+  def this(errorClass: String,
+           errorSubClass: String,
+           messageParameters: Array[String],
+           ctx: ParserRuleContext) =
+    this(Option(ParserUtils.command(ctx)),
+      SparkThrowableHelper.getMessage(errorClass, errorSubClass, messageParameters),
+      ParserUtils.position(ctx.getStart),
+      ParserUtils.position(ctx.getStop),
+      Some(errorClass),
+      Some(errorSubClass),
       messageParameters)
 
   /** Compose the message through SparkThrowableHelper given errorClass and messageParameters. */
@@ -271,10 +286,11 @@ class ParseException(
       messageParameters: Array[String]) =
     this(
       command,
-      SparkThrowableHelper.getMessage(errorClass, messageParameters),
+      SparkThrowableHelper.getMessage(errorClass, null, messageParameters),
       start,
       stop,
       Some(errorClass),
+      None,
       messageParameters)
 
   override def getMessage: String = {
@@ -303,7 +319,7 @@ class ParseException(
     if (cmd.trim().isEmpty && errorClass.isDefined && errorClass.get == "PARSE_SYNTAX_ERROR") {
       new ParseException(Option(cmd), start, stop, "PARSE_EMPTY_STATEMENT", Array[String]())
     } else {
-      new ParseException(Option(cmd), message, start, stop, errorClass, messageParameters)
+      new ParseException(Option(cmd), message, start, stop, errorClass, None, messageParameters)
     }
   }
 }
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index afd5cb2a073d4..551eaa6aeb7da 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -95,8 +95,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
   def unsupportedIfNotExistsError(tableName: String): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("INSERT_PARTITION_SPEC_IF_NOT_EXISTS",
-        toSQLId(tableName)))
+      errorSubClass = "INSERT_PARTITION_SPEC_IF_NOT_EXISTS",
+      messageParameters = Array(toSQLId(tableName)))
   }
 
   def nonPartitionColError(partitionName: String): Throwable = {
@@ -113,18 +113,21 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def nestedGeneratorError(trimmedNestedGenerator: Expression): Throwable = {
     new AnalysisException(errorClass = "UNSUPPORTED_GENERATOR",
-      messageParameters = Array("NESTED_IN_EXPRESSIONS", toSQLExpr(trimmedNestedGenerator)))
+      errorSubClass = "NESTED_IN_EXPRESSIONS",
+      messageParameters = Array(toSQLExpr(trimmedNestedGenerator)))
   }
 
   def moreThanOneGeneratorError(generators: Seq[Expression], clause: String): Throwable = {
     new AnalysisException(errorClass = "UNSUPPORTED_GENERATOR",
-      messageParameters = Array("MULTI_GENERATOR",
-        clause, generators.size.toString, generators.map(toSQLExpr).mkString(", ")))
+      errorSubClass = "MULTI_GENERATOR",
+      messageParameters = Array(clause,
+        generators.size.toString, generators.map(toSQLExpr).mkString(", ")))
   }
 
   def generatorOutsideSelectError(plan: LogicalPlan): Throwable = {
     new AnalysisException(errorClass = "UNSUPPORTED_GENERATOR",
-      messageParameters = Array("OUTSIDE_SELECT", plan.simpleString(SQLConf.get.maxToStringFields)))
+      errorSubClass = "OUTSIDE_SELECT",
+      messageParameters = Array(plan.simpleString(SQLConf.get.maxToStringFields)))
   }
 
   def legacyStoreAssignmentPolicyError(): Throwable = {
@@ -143,19 +146,20 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def dataTypeMismatchForDeserializerError(
       dataType: DataType, desiredType: String): Throwable = {
-    val quantifier = if (desiredType.equals("array")) "an" else "a"
     new AnalysisException(
       errorClass = "UNSUPPORTED_DESERIALIZER",
+      errorSubClass = "DATA_TYPE_MISMATCH",
       messageParameters =
-        Array("DATA_TYPE_MISMATCH", quantifier, toSQLType(desiredType), toSQLType(dataType)))
+        Array(toSQLType(desiredType), toSQLType(dataType)))
   }
 
   def fieldNumberMismatchForDeserializerError(
       schema: StructType, maxOrdinal: Int): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_DESERIALIZER",
+      errorSubClass = "FIELD_NUMBER_MISMATCH",
       messageParameters =
-        Array("FIELD_NUMBER_MISMATCH", toSQLType(schema), (maxOrdinal + 1).toString))
+        Array(toSQLType(schema), (maxOrdinal + 1).toString))
   }
 
   def upCastFailureError(
@@ -203,7 +207,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
   def pandasUDFAggregateNotSupportedInPivotError(): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("PANDAS_UDAF_IN_PIVOT"))
+      errorSubClass = "PANDAS_UDAF_IN_PIVOT",
+        messageParameters = Array[String]())
   }
 
   def aggregateExpressionRequiredForPivotError(sql: String): Throwable = {
@@ -323,7 +328,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def generatorNotExpectedError(name: FunctionIdentifier, classCanonicalName: String): Throwable = {
     new AnalysisException(errorClass = "UNSUPPORTED_GENERATOR",
-      messageParameters = Array("NOT_GENERATOR", toSQLId(name.toString), classCanonicalName))
+      errorSubClass = "NOT_GENERATOR",
+      messageParameters = Array(toSQLId(name.toString), classCanonicalName))
   }
 
   def functionWithUnsupportedSyntaxError(prettyName: String, syntax: String): Throwable = {
@@ -1530,7 +1536,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def secondArgumentOfFunctionIsNotIntegerError(
       function: String, e: NumberFormatException): Throwable = {
-    // The second argument of '{function}' function needs to be an integer
+    // The second argument of {function} function needs to be an integer
     new AnalysisException(
       errorClass = "SECOND_FUNCTION_ARGUMENT_NOT_INTEGER",
       messageParameters = Array(function),
@@ -1585,7 +1591,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
   def usePythonUDFInJoinConditionUnsupportedError(joinType: JoinType): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("PYTHON_UDF_IN_ON_CLAUSE", s"${toSQLStmt(joinType.sql)}"))
+      errorSubClass = "PYTHON_UDF_IN_ON_CLAUSE",
+      messageParameters = Array(s"${toSQLStmt(joinType.sql)}"))
   }
 
   def conflictingAttributesInJoinConditionError(
@@ -1946,14 +1953,14 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
     new AnalysisException(
       errorClass = "FORBIDDEN_OPERATION",
       messageParameters =
-        Array(toSQLStmt("DESC PARTITION"), "the temporary view", toSQLId(table)))
+        Array(toSQLStmt("DESC PARTITION"), "TEMPORARY VIEW", toSQLId(table)))
   }
 
   def descPartitionNotAllowedOnView(table: String): Throwable = {
     new AnalysisException(
       errorClass = "FORBIDDEN_OPERATION",
       messageParameters = Array(
-        toSQLStmt("DESC PARTITION"), "the view", toSQLId(table)))
+        toSQLStmt("DESC PARTITION"), "VIEW", toSQLId(table)))
   }
 
   def showPartitionNotAllowedOnTableNotPartitionedError(tableIdentWithDB: String): Throwable = {
@@ -2331,7 +2338,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
   def udfClassWithTooManyTypeArgumentsError(n: Int): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("TOO_MANY_TYPE_ARGUMENTS_FOR_UDF_CLASS", s"$n"))
+      errorSubClass = "TOO_MANY_TYPE_ARGUMENTS_FOR_UDF_CLASS",
+      messageParameters = Array(s"$n"))
   }
 
   def classWithoutPublicNonArgumentConstructorError(className: String): Throwable = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index c8feb4c070257..cd258e3649a75 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -226,6 +226,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def invalidFractionOfSecondError(): DateTimeException = {
     new SparkDateTimeException(
       errorClass = "INVALID_FRACTION_OF_SECOND",
+      errorSubClass = None,
       Array(toSQLConf(SQLConf.ANSI_ENABLED.key)))
   }
 
@@ -274,14 +275,15 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def literalTypeUnsupportedError(v: Any): RuntimeException = {
     new SparkRuntimeException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("LITERAL_TYPE", s"${v.toString}", s"${v.getClass.toString}"))
+      errorSubClass = "LITERAL_TYPE",
+      messageParameters = Array( s"${v.toString}", s"${v.getClass.toString}"))
   }
 
   def pivotColumnUnsupportedError(v: Any, dataType: DataType): RuntimeException = {
     new SparkRuntimeException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("PIVOT_TYPE",
-        s"${v.toString}", s"${toSQLType(dataType)}"))
+      errorSubClass = "PIVOT_TYPE",
+      messageParameters = Array(s"${v.toString}", s"${toSQLType(dataType)}"))
   }
 
   def noDefaultForDataTypeError(dataType: DataType): RuntimeException = {
@@ -555,15 +557,16 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   }
 
   def incompatibleDataSourceRegisterError(e: Throwable): Throwable = {
-    new SparkClassNotFoundException("INCOMPATIBLE_DATASOURCE_REGISTER", Array(e.getMessage), e)
+    new SparkClassNotFoundException("INCOMPATIBLE_DATASOURCE_REGISTER", None,
+      Array(e.getMessage), e)
   }
 
   def sparkUpgradeInReadingDatesError(
       format: String, config: String, option: String): SparkUpgradeException = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
+      errorSubClass = Some("READ_ANCIENT_DATETIME"),
       messageParameters = Array(
-        "READ_ANCIENT_DATETIME",
         format,
         toSQLConf(config),
         toDSOption(option),
@@ -576,8 +579,8 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def sparkUpgradeInWritingDatesError(format: String, config: String): SparkUpgradeException = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
+      errorSubClass = Some("WRITE_ANCIENT_DATETIME"),
       messageParameters = Array(
-        "WRITE_ANCIENT_DATETIME",
         format,
         toSQLConf(config),
         toSQLConf(config)),
@@ -610,9 +613,11 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def saveModeUnsupportedError(saveMode: Any, pathExists: Boolean): Throwable = {
     pathExists match {
       case true => new SparkIllegalArgumentException(errorClass = "UNSUPPORTED_SAVE_MODE",
-        messageParameters = Array("EXISTENT_PATH", toSQLValue(saveMode, StringType)))
+        errorSubClass = Some("EXISTENT_PATH"),
+        messageParameters = Array(toSQLValue(saveMode, StringType)))
       case _ => new SparkIllegalArgumentException(errorClass = "UNSUPPORTED_SAVE_MODE",
-        messageParameters = Array("NON_EXISTENT_PATH", toSQLValue(saveMode, StringType)))
+        errorSubClass = Some("NON_EXISTENT_PATH"),
+        messageParameters = Array(toSQLValue(saveMode, StringType)))
     }
   }
 
@@ -796,7 +801,8 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def transactionUnsupportedByJdbcServerError(): Throwable = {
     new SparkSQLFeatureNotSupportedException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("JDBC_TRANSACTION"))
+      errorSubClass = "JDBC_TRANSACTION",
+      messageParameters = Array[String]())
   }
 
   def dataTypeUnsupportedYetError(dataType: DataType): Throwable = {
@@ -1014,8 +1020,8 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def failToParseDateTimeInNewParserError(s: String, e: Throwable): Throwable = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
+      errorSubClass = Some("PARSE_DATETIME_BY_NEW_PARSER"),
       messageParameters = Array(
-        "PARSE_DATETIME_BY_NEW_PARSER",
         toSQLValue(s, StringType),
         toSQLConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key)),
       e)
@@ -1025,8 +1031,8 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
       resultCandidate: String, e: Throwable): Throwable = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
+      errorSubClass = Some("PARSE_DATETIME_BY_NEW_PARSER"),
       messageParameters = Array(
-        "PARSE_DATETIME_BY_NEW_PARSER",
         toSQLValue(resultCandidate, StringType),
         toSQLConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key)),
       e)
@@ -1035,8 +1041,8 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def failToRecognizePatternAfterUpgradeError(pattern: String, e: Throwable): Throwable = {
     new SparkUpgradeException(
       errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
+      errorSubClass = Some("DATETIME_PATTERN_RECOGNITION"),
       messageParameters = Array(
-        "DATETIME_PATTERN_RECOGNITION",
         toSQLValue(pattern, StringType),
         toSQLConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key)),
       e)
@@ -1054,7 +1060,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   }
 
   def concurrentQueryInstanceError(): Throwable = {
-    new SparkConcurrentModificationException("CONCURRENT_QUERY", Array.empty)
+    new SparkConcurrentModificationException("CONCURRENT_QUERY", None, Array.empty)
   }
 
   def cannotParseJsonArraysAsStructsError(): Throwable = {
@@ -1333,7 +1339,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
 
   def indexOutOfBoundsOfArrayDataError(idx: Int): Throwable = {
     new SparkIndexOutOfBoundsException(
-      errorClass = "INDEX_OUT_OF_BOUNDS", Array(toSQLValue(idx, IntegerType)))
+      errorClass = "INDEX_OUT_OF_BOUNDS", None, Array(toSQLValue(idx, IntegerType)))
   }
 
   def malformedRecordsDetectedInRecordParsingError(e: BadRecordException): Throwable = {
@@ -1454,7 +1460,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   }
 
   def renamePathAsExistsPathError(srcPath: Path, dstPath: Path): Throwable = {
-    new SparkFileAlreadyExistsException(errorClass = "FAILED_RENAME_PATH",
+    new SparkFileAlreadyExistsException(errorClass = "FAILED_RENAME_PATH", None,
       Array(srcPath.toString, dstPath.toString))
   }
 
@@ -1463,7 +1469,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   }
 
   def renameSrcPathNotFoundError(srcPath: Path): Throwable = {
-    new SparkFileNotFoundException(errorClass = "RENAME_SRC_PATH_NOT_FOUND",
+    new SparkFileNotFoundException(errorClass = "RENAME_SRC_PATH_NOT_FOUND", None,
       Array(srcPath.toString))
   }
 
@@ -1661,7 +1667,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
       permission: FsPermission,
       path: Path,
       e: Throwable): Throwable = {
-    new SparkSecurityException(errorClass = "RESET_PERMISSION_TO_ORIGINAL",
+    new SparkSecurityException(errorClass = "RESET_PERMISSION_TO_ORIGINAL", None,
       Array(permission.toString, path.toString, e.getMessage))
   }
 
@@ -1908,13 +1914,15 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def repeatedPivotsUnsupportedError(): Throwable = {
     new SparkUnsupportedOperationException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("REPEATED_PIVOT"))
+      errorSubClass = "REPEATED_PIVOT",
+      messageParameters = Array[String]())
   }
 
   def pivotNotAfterGroupByUnsupportedError(): Throwable = {
     new SparkUnsupportedOperationException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("PIVOT_AFTER_GROUP_BY"))
+      errorSubClass = "PIVOT_AFTER_GROUP_BY",
+      messageParameters = Array[String]())
   }
 
   private val aesFuncName = toSQLId("aes_encrypt") + "/" + toSQLId("aes_decrypt")
@@ -1924,14 +1932,15 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
       errorClass = "INVALID_PARAMETER_VALUE",
       messageParameters = Array(
         "key",
-        s"the $aesFuncName function",
+        aesFuncName,
         s"expects a binary value with 16, 24 or 32 bytes, but got ${actualLength.toString} bytes."))
   }
 
   def aesModeUnsupportedError(mode: String, padding: String): RuntimeException = {
     new SparkRuntimeException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("AES_MODE", mode, padding, aesFuncName))
+      errorSubClass = "AES_MODE",
+      messageParameters = Array(mode, padding, aesFuncName))
   }
 
   def aesCryptoError(detailMessage: String): RuntimeException = {
@@ -1939,7 +1948,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
       errorClass = "INVALID_PARAMETER_VALUE",
       messageParameters = Array(
         "expr, key",
-        s"the $aesFuncName function",
+        aesFuncName,
         s"Detail message: $detailMessage"))
   }
 
@@ -1950,16 +1959,16 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
   def cannotConvertOrcTimestampToTimestampNTZError(): Throwable = {
     new SparkUnsupportedOperationException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("ORC_TYPE_CAST",
-        toSQLType(TimestampType),
+      errorSubClass = "ORC_TYPE_CAST",
+      messageParameters = Array(toSQLType(TimestampType),
         toSQLType(TimestampNTZType)))
   }
 
   def cannotConvertOrcTimestampNTZToTimestampLTZError(): Throwable = {
     new SparkUnsupportedOperationException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("ORC_TYPE_CAST",
-        toSQLType(TimestampNTZType),
+      errorSubClass = "ORC_TYPE_CAST",
+      messageParameters = Array(toSQLType(TimestampNTZType),
         toSQLType(TimestampType)))
   }
 
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
index b7037cdaeb139..d4629f0dd3fe4 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala
@@ -96,14 +96,16 @@ private[sql] object QueryParsingErrors extends QueryErrorsBase {
   def transformNotSupportQuantifierError(ctx: ParserRuleContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("TRANSFORM_DISTINCT_ALL"),
+      errorSubClass = "TRANSFORM_DISTINCT_ALL",
+        messageParameters = Array[String](),
       ctx)
   }
 
   def transformWithSerdeUnsupportedError(ctx: ParserRuleContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("TRANSFORM_NON_HIVE"),
+      errorSubClass = "TRANSFORM_NON_HIVE",
+        messageParameters = Array[String](),
       ctx)
   }
 
@@ -114,21 +116,24 @@ private[sql] object QueryParsingErrors extends QueryErrorsBase {
   def lateralJoinWithNaturalJoinUnsupportedError(ctx: ParserRuleContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("LATERAL_NATURAL_JOIN"),
+      errorSubClass = "LATERAL_NATURAL_JOIN",
+        messageParameters = Array[String](),
       ctx)
   }
 
   def lateralJoinWithUsingJoinUnsupportedError(ctx: ParserRuleContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("LATERAL_JOIN_USING"),
+      errorSubClass = "LATERAL_JOIN_USING",
+        messageParameters = Array[String](),
       ctx)
   }
 
   def unsupportedLateralJoinTypeError(ctx: ParserRuleContext, joinType: String): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("LATERAL_JOIN_OF_TYPE", s"${toSQLStmt(joinType)}"),
+      errorSubClass = "LATERAL_JOIN_OF_TYPE",
+      messageParameters = Array(s"${toSQLStmt(joinType)}"),
       ctx)
   }
 
@@ -155,7 +160,10 @@ private[sql] object QueryParsingErrors extends QueryErrorsBase {
   }
 
   def naturalCrossJoinUnsupportedError(ctx: RelationContext): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE", Array("NATURAL_CROSS_JOIN"), ctx)
+    new ParseException(errorClass = "UNSUPPORTED_FEATURE",
+      errorSubClass = "NATURAL_CROSS_JOIN",
+        messageParameters = Array[String](),
+      ctx = ctx)
   }
 
   def emptyInputForTableSampleError(ctx: ParserRuleContext): Throwable = {
@@ -269,14 +277,16 @@ private[sql] object QueryParsingErrors extends QueryErrorsBase {
       property: String, ctx: ParserRuleContext, msg: String): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("SET_NAMESPACE_PROPERTY", property, msg),
+      errorSubClass = "SET_NAMESPACE_PROPERTY",
+      messageParameters = Array(property, msg),
       ctx)
   }
 
   def propertiesAndDbPropertiesBothSpecifiedError(ctx: CreateNamespaceContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("SET_PROPERTIES_AND_DBPROPERTIES"),
+      errorSubClass = "SET_PROPERTIES_AND_DBPROPERTIES",
+        messageParameters = Array[String](),
       ctx
     )
   }
@@ -285,7 +295,8 @@ private[sql] object QueryParsingErrors extends QueryErrorsBase {
       property: String, ctx: ParserRuleContext, msg: String): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("SET_TABLE_PROPERTY", property, msg),
+      errorSubClass = "SET_TABLE_PROPERTY",
+      messageParameters = Array(property, msg),
       ctx)
   }
 
@@ -318,7 +329,8 @@ private[sql] object QueryParsingErrors extends QueryErrorsBase {
   def descColumnForPartitionUnsupportedError(ctx: DescribeRelationContext): Throwable = {
     new ParseException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("DESC_TABLE_COLUMN_PARTITION"),
+      errorSubClass = "DESC_TABLE_COLUMN_PARTITION",
+      messageParameters = Array[String](),
       ctx)
   }
 
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala
index e5a3a531059eb..c0877bea14809 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala
@@ -86,15 +86,19 @@ class EncoderResolutionSuite extends PlanTest {
   test("the real type is not compatible with encoder schema: primitive array") {
     val encoder = ExpressionEncoder[PrimitiveArrayClass]
     val attrs = Seq($"arr".array(StringType))
-    assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-      s"""
-         |[CANNOT_UP_CAST_DATATYPE] Cannot up cast array element from "STRING" to "BIGINT".
-         |The type path of the target object is:
-         |- array element class: "scala.Long"
-         |- field (class: "scala.Array", name: "arr")
-         |- root class: "org.apache.spark.sql.catalyst.encoders.PrimitiveArrayClass"
-         |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")
+    checkError(
+      exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+      errorClass = "CANNOT_UP_CAST_DATATYPE",
+      parameters = Map("expression" -> "array element",
+        "sourceType" -> "\"STRING\"", "targetType" -> "\"BIGINT\"",
+        "details" -> (
+          s"""
+          |The type path of the target object is:
+          |- array element class: "scala.Long"
+          |- field (class: "scala.Array", name: "arr")
+          |- root class: "org.apache.spark.sql.catalyst.encoders.PrimitiveArrayClass"
+          |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("real type doesn't match encoder schema but they are compatible: array") {
@@ -117,9 +121,11 @@ class EncoderResolutionSuite extends PlanTest {
   test("the real type is not compatible with encoder schema: non-array field") {
     val encoder = ExpressionEncoder[ArrayClass]
     val attrs = Seq($"arr".int)
-    assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-      """[UNSUPPORTED_DESERIALIZER.DATA_TYPE_MISMATCH] """ +
-      """The deserializer is not supported: need an "ARRAY" field but got "INT".""")
+    checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+      errorClass = "UNSUPPORTED_DESERIALIZER",
+      errorSubClass = Some("DATA_TYPE_MISMATCH"),
+      parameters = Map("desiredType" -> "\"ARRAY\"", "dataType" -> "\"INT\""),
+      sqlState = None)
   }
 
   test("the real type is not compatible with encoder schema: array element type") {
@@ -134,9 +140,11 @@ class EncoderResolutionSuite extends PlanTest {
 
     withClue("inner element is not array") {
       val attrs = Seq($"nestedArr".array(new StructType().add("arr", "int")))
-      assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-        """[UNSUPPORTED_DESERIALIZER.DATA_TYPE_MISMATCH] """ +
-        """The deserializer is not supported: need an "ARRAY" field but got "INT".""")
+      checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+        errorClass = "UNSUPPORTED_DESERIALIZER",
+        errorSubClass = Some("DATA_TYPE_MISMATCH"),
+        parameters = Map("desiredType" -> "\"ARRAY\"", "dataType" -> "\"INT\""),
+        sqlState = None)
     }
 
     withClue("nested array element type is not compatible") {
@@ -169,18 +177,22 @@ class EncoderResolutionSuite extends PlanTest {
 
     {
       val attrs = Seq($"a".string, $"b".long, $"c".int)
-      assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-        """[UNSUPPORTED_DESERIALIZER.FIELD_NUMBER_MISMATCH] The deserializer is not supported: """ +
-        """try to map "STRUCT<a: STRING, b: BIGINT, c: INT>" to Tuple2, """ +
-        """but failed as the number of fields does not line up.""")
+      checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+        errorClass = "UNSUPPORTED_DESERIALIZER",
+        errorSubClass = Some("FIELD_NUMBER_MISMATCH"),
+        parameters = Map("schema" -> "\"STRUCT<a: STRING, b: BIGINT, c: INT>\"",
+          "ordinal" -> "2"),
+        sqlState = None)
     }
 
     {
       val attrs = Seq($"a".string)
-      assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-        """[UNSUPPORTED_DESERIALIZER.FIELD_NUMBER_MISMATCH] """ +
-        """The deserializer is not supported: try to map "STRUCT<a: STRING>" to Tuple2, """ +
-        """but failed as the number of fields does not line up.""")
+      checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+        errorClass = "UNSUPPORTED_DESERIALIZER",
+        errorSubClass = Some("FIELD_NUMBER_MISMATCH"),
+        parameters = Map("schema" -> "\"STRUCT<a: STRING>\"",
+          "ordinal" -> "2"),
+        sqlState = None)
     }
   }
 
@@ -189,18 +201,22 @@ class EncoderResolutionSuite extends PlanTest {
 
     {
       val attrs = Seq($"a".string, $"b".struct($"x".long, $"y".string, $"z".int))
-      assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-        """[UNSUPPORTED_DESERIALIZER.FIELD_NUMBER_MISMATCH] The deserializer is not supported: """ +
-        """try to map "STRUCT<x: BIGINT, y: STRING, z: INT>" to Tuple2, """ +
-        """but failed as the number of fields does not line up.""")
+      checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+        errorClass = "UNSUPPORTED_DESERIALIZER",
+        errorSubClass = Some("FIELD_NUMBER_MISMATCH"),
+        parameters = Map("schema" -> "\"STRUCT<x: BIGINT, y: STRING, z: INT>\"",
+          "ordinal" -> "2"),
+        sqlState = None)
     }
 
     {
       val attrs = Seq($"a".string, $"b".struct($"x".long))
-      assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-        """[UNSUPPORTED_DESERIALIZER.FIELD_NUMBER_MISMATCH] The deserializer is not supported: """ +
-        """try to map "STRUCT<x: BIGINT>" to Tuple2, """ +
-        """but failed as the number of fields does not line up.""")
+      checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+        errorClass = "UNSUPPORTED_DESERIALIZER",
+        errorSubClass = Some("FIELD_NUMBER_MISMATCH"),
+        parameters = Map("schema" -> "\"STRUCT<x: BIGINT>\"",
+          "ordinal" -> "2"),
+        sqlState = None)
     }
   }
 
@@ -216,42 +232,52 @@ class EncoderResolutionSuite extends PlanTest {
     Seq($"a".struct($"x".long), $"a".array(StringType), Symbol("a").map(StringType, StringType))
       .foreach { attr =>
         val attrs = Seq(attr)
-        assert(intercept[AnalysisException](encoder.resolveAndBind(attrs)).message ==
-        s"""
-           |[CANNOT_UP_CAST_DATATYPE] Cannot up cast a from "${attr.dataType.sql}" to "STRING".
-           |The type path of the target object is:
-           |- root class: "java.lang.String"
-           |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")
+        checkError(exception = intercept[AnalysisException](encoder.resolveAndBind(attrs)),
+          errorClass = "CANNOT_UP_CAST_DATATYPE",
+          parameters = Map("expression" -> "a",
+            "sourceType" -> ("\"" + attr.dataType.sql + "\""), "targetType" -> "\"STRING\"",
+            "details" -> (
+          s"""
+          |The type path of the target object is:
+          |- root class: "java.lang.String"
+          |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("throw exception if real type is not compatible with encoder schema") {
-    val msg1 = intercept[AnalysisException] {
+    val e1 = intercept[AnalysisException] {
       ExpressionEncoder[StringIntClass].resolveAndBind(Seq($"a".string, $"b".long))
-    }.message
-    assert(msg1 ==
-      s"""
-         |[CANNOT_UP_CAST_DATATYPE] 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.catalyst.encoders.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] {
+    }
+    checkError(exception = e1,
+      errorClass = "CANNOT_UP_CAST_DATATYPE",
+      parameters = Map("expression" -> "b",
+        "sourceType" -> ("\"BIGINT\""), "targetType" -> "\"INT\"",
+        "details" -> (
+          s"""
+          |The type path of the target object is:
+          |- field (class: "scala.Int", name: "b")
+          |- root class: "org.apache.spark.sql.catalyst.encoders.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 e2 = intercept[AnalysisException] {
       val structType = new StructType().add("a", StringType).add("b", DecimalType.SYSTEM_DEFAULT)
       ExpressionEncoder[ComplexClass].resolveAndBind(Seq($"a".long, $"b".struct(structType)))
-    }.message
-    assert(msg2 ==
-      s"""
-         |[CANNOT_UP_CAST_DATATYPE] 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.catalyst.encoders.StringLongClass", name: "b")
-         |- root class: "org.apache.spark.sql.catalyst.encoders.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")
+    }
+
+    checkError(exception = e2,
+      errorClass = "CANNOT_UP_CAST_DATATYPE",
+      parameters = Map("expression" -> "b.`b`",
+        "sourceType" -> ("\"DECIMAL(38,18)\""), "targetType" -> "\"BIGINT\"",
+        "details" -> (
+          s"""
+          |The type path of the target object is:
+          |- field (class: "scala.Long", name: "b")
+          |- field (class: "org.apache.spark.sql.catalyst.encoders.StringLongClass", name: "b")
+          |- root class: "org.apache.spark.sql.catalyst.encoders.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("SPARK-31750: eliminate UpCast if child's dataType is DecimalType") {
diff --git a/sql/core/src/test/resources/sql-tests/results/date.sql.out b/sql/core/src/test/resources/sql-tests/results/date.sql.out
index e69bf93942e08..f5df5f7577d04 100644
--- a/sql/core/src/test/resources/sql-tests/results/date.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/date.sql.out
@@ -319,7 +319,7 @@ select date_add('2011-11-11', '1.2')
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of 'date_add' function needs to be an integer.
+[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of date_add function needs to be an integer.
 
 
 -- !query
@@ -427,7 +427,7 @@ select date_sub(date'2011-11-11', '1.2')
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of 'date_sub' function needs to be an integer.
+[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of date_sub function needs to be an integer.
 
 
 -- !query
diff --git a/sql/core/src/test/resources/sql-tests/results/datetime-legacy.sql.out b/sql/core/src/test/resources/sql-tests/results/datetime-legacy.sql.out
index 54351adf20059..08c64b357398c 100644
--- a/sql/core/src/test/resources/sql-tests/results/datetime-legacy.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/datetime-legacy.sql.out
@@ -319,7 +319,7 @@ select date_add('2011-11-11', '1.2')
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of 'date_add' function needs to be an integer.
+[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of date_add function needs to be an integer.
 
 
 -- !query
@@ -427,7 +427,7 @@ select date_sub(date'2011-11-11', '1.2')
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of 'date_sub' function needs to be an integer.
+[SECOND_FUNCTION_ARGUMENT_NOT_INTEGER] The second argument of date_sub function needs to be an integer.
 
 
 -- !query
diff --git a/sql/core/src/test/resources/sql-tests/results/describe.sql.out b/sql/core/src/test/resources/sql-tests/results/describe.sql.out
index 7c0cc29a814a0..8a6471bf2399d 100644
--- a/sql/core/src/test/resources/sql-tests/results/describe.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/describe.sql.out
@@ -462,7 +462,7 @@ DESC temp_v PARTITION (c='Us', d=1)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-[FORBIDDEN_OPERATION] The operation DESC PARTITION is not allowed on the temporary view: `temp_v`
+[FORBIDDEN_OPERATION] The operation DESC PARTITION is not allowed on the TEMPORARY VIEW: `temp_v`
 
 
 -- !query
@@ -539,7 +539,7 @@ DESC v PARTITION (c='Us', d=1)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-[FORBIDDEN_OPERATION] The operation DESC PARTITION is not allowed on the view: `v`
+[FORBIDDEN_OPERATION] The operation DESC PARTITION is not allowed on the VIEW: `v`
 
 
 -- !query
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsDSv2Suite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsDSv2Suite.scala
index 8d9d8e277351c..4a847ca0340ec 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsDSv2Suite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsDSv2Suite.scala
@@ -50,13 +50,12 @@ class QueryCompilationErrorsDSv2Suite
         }
 
         checkAnswer(spark.table(tbl), spark.emptyDataFrame)
-        checkErrorClass(
+        checkError(
           exception = e,
           errorClass = "UNSUPPORTED_FEATURE",
-          errorSubClass = Some("INSERT_PARTITION_SPEC_IF_NOT_EXISTS"),
-          msg = "The feature is not supported: " +
-            s"""INSERT INTO `testcat`.`ns1`.`ns2`.`tbl` IF NOT EXISTS in the PARTITION spec.""",
-          sqlState = Some("0A000"))
+          errorSubClass = "INSERT_PARTITION_SPEC_IF_NOT_EXISTS",
+          parameters = Map("tableName" -> "`testcat`.`ns1`.`ns2`.`tbl`"),
+          sqlState = "0A000")
       }
     }
   }
@@ -71,10 +70,10 @@ class QueryCompilationErrorsDSv2Suite
       }
 
       verifyTable(t1, spark.emptyDataFrame)
-      checkErrorClass(
+      checkError(
         exception = e,
         errorClass = "NON_PARTITION_COLUMN",
-        msg = "PARTITION clause cannot contain the non-partition column: `id`.")
+        parameters = Map("columnName" -> "`id`"))
     }
   }
 
@@ -88,10 +87,10 @@ class QueryCompilationErrorsDSv2Suite
       }
 
       verifyTable(t1, spark.emptyDataFrame)
-      checkErrorClass(
+      checkError(
         exception = e,
         errorClass = "NON_PARTITION_COLUMN",
-        msg = "PARTITION clause cannot contain the non-partition column: `data`.")
+        parameters = Map("columnName" -> "`data`"))
     }
   }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
index 4a440dc6ab719..4f631292436c9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala
@@ -41,35 +41,36 @@ class QueryCompilationErrorsSuite
     val e1 = intercept[AnalysisException] {
       sql("select 'value1' as a, 1L as b").as[StringIntClass]
     }
-    checkErrorClass(
+    checkError(
       exception = e1,
       errorClass = "CANNOT_UP_CAST_DATATYPE",
-      msg =
+      parameters = Map("expression" -> "b", "sourceType" -> "\"BIGINT\"", "targetType" -> "\"INT\"",
+        "details" -> (
         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")
+         """.stripMargin.trim + " of the field in the target object")))
 
     val e2 = intercept[AnalysisException] {
       sql("select 1L as a," +
         " named_struct('a', 'value1', 'b', cast(1.0 as decimal(38,18))) as b")
         .as[ComplexClass]
     }
-    checkErrorClass(
+    checkError(
       exception = e2,
       errorClass = "CANNOT_UP_CAST_DATATYPE",
-      msg =
+      parameters = Map("expression" -> "b.`b`", "sourceType" -> "\"DECIMAL(38,18)\"",
+        "targetType" -> "\"BIGINT\"",
+        "details" -> (
         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")
+         """.stripMargin.trim + " of the field in the target object")))
   }
 
   test("UNSUPPORTED_GROUPING_EXPRESSION: filter with grouping/grouping_Id expression") {
@@ -83,10 +84,10 @@ class QueryCompilationErrorsSuite
         df.groupBy("CustomerId").agg(Map("Quantity" -> "max"))
           .filter(s"$grouping(CustomerId)=17850")
       }
-      checkErrorClass(
+      checkError(
         exception = e,
         errorClass = "UNSUPPORTED_GROUPING_EXPRESSION",
-        msg = "grouping()/grouping_id() can only be used with GroupingSets/Cube/Rollup")
+        parameters = Map[String, String]())
     }
   }
 
@@ -101,10 +102,10 @@ class QueryCompilationErrorsSuite
         df.groupBy("CustomerId").agg(Map("Quantity" -> "max")).
           sort(grouping)
       }
-      checkErrorClass(
+      checkError(
         exception = e,
         errorClass = "UNSUPPORTED_GROUPING_EXPRESSION",
-        msg = "grouping()/grouping_id() can only be used with GroupingSets/Cube/Rollup")
+        parameters = Map[String, String]())
     }
   }
 
@@ -138,11 +139,10 @@ class QueryCompilationErrorsSuite
         .collect()
     }
 
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "INVALID_PANDAS_UDF_PLACEMENT",
-      msg = "The group aggregate pandas UDF `pandas_udf_1`, `pandas_udf_2` cannot be invoked " +
-        "together with as other, non-pandas aggregate functions.")
+      parameters = Map("functionList" -> "`pandas_udf_1`, `pandas_udf_2`"))
   }
 
   test("UNSUPPORTED_FEATURE: Using Python UDF with unsupported join condition") {
@@ -165,12 +165,11 @@ class QueryCompilationErrorsSuite
         df2, pythonTestUDF(df1("CustomerID") === df2("CustomerID")), "leftouter").collect()
     }
 
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_FEATURE",
       errorSubClass = Some("PYTHON_UDF_IN_ON_CLAUSE"),
-      msg = "The feature is not supported: " +
-        "Python UDF in the ON clause of a LEFT OUTER JOIN.",
+      parameters = Map("joinType" -> "LEFT OUTER"),
       sqlState = Some("0A000"))
   }
 
@@ -189,13 +188,12 @@ class QueryCompilationErrorsSuite
       df.groupBy(df("CustomerID")).pivot(df("CustomerID")).agg(pandasTestUDF(df("Quantity")))
     }
 
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_FEATURE",
-      errorSubClass = Some("PANDAS_UDAF_IN_PIVOT"),
-      msg = "The feature is not supported: " +
-        "Pandas user defined aggregate function in the PIVOT clause.",
-      sqlState = Some("0A000"))
+      errorSubClass = "PANDAS_UDAF_IN_PIVOT",
+      parameters = Map[String, String](),
+      sqlState = "0A000")
   }
 
   test("NO_HANDLER_FOR_UDAF: No handler for UDAF error") {
@@ -219,21 +217,10 @@ class QueryCompilationErrorsSuite
   }
 
   test("UNTYPED_SCALA_UDF: use untyped Scala UDF should fail by default") {
-    checkErrorClass(
+    checkError(
       exception = intercept[AnalysisException](udf((x: Int) => x, IntegerType)),
       errorClass = "UNTYPED_SCALA_UDF",
-      msg =
-        "You're using untyped Scala UDF, which does not have the input type " +
-        "information. Spark may blindly pass null to the Scala closure with primitive-type " +
-        "argument, and the closure will see the default value of the Java type for the null " +
-        "argument, e.g. `udf((x: Int) => x, IntegerType)`, the result is 0 for null input. " +
-        "To get rid of this error, you could:\n" +
-        "1. use typed Scala UDF APIs(without return type parameter), e.g. `udf((x: Int) => x)`\n" +
-        "2. use Java UDF APIs, e.g. `udf(new UDF1[String, Integer] { " +
-        "override def call(s: String): Integer = s.length() }, IntegerType)`, " +
-        "if input types are all non primitive\n" +
-        s"""3. set "${SQLConf.LEGACY_ALLOW_UNTYPED_SCALA_UDF.key}" to "true" and """ +
-        s"use this API with caution")
+      parameters = Map[String, String]())
   }
 
   test("NO_UDF_INTERFACE_ERROR: java udf class does not implement any udf interface") {
@@ -244,10 +231,10 @@ class QueryCompilationErrorsSuite
         className,
         StringType)
     )
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "NO_UDF_INTERFACE_ERROR",
-      msg = s"UDF class $className doesn't implement any UDF interface")
+      parameters = Map("className" -> className))
   }
 
   test("MULTI_UDF_INTERFACE_ERROR: java udf implement multi UDF interface") {
@@ -258,10 +245,10 @@ class QueryCompilationErrorsSuite
         className,
         StringType)
     )
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "MULTI_UDF_INTERFACE_ERROR",
-      msg = s"Not allowed to implement multiple UDF interfaces, UDF class $className")
+      parameters = Map("className" -> className))
   }
 
   test("UNSUPPORTED_FEATURE: java udf with too many type arguments") {
@@ -272,38 +259,39 @@ class QueryCompilationErrorsSuite
         className,
         StringType)
     )
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_FEATURE",
-      errorSubClass = Some("TOO_MANY_TYPE_ARGUMENTS_FOR_UDF_CLASS"),
-      msg = "The feature is not supported: UDF class with 24 type arguments.",
-      sqlState = Some("0A000"))
+      errorSubClass = "TOO_MANY_TYPE_ARGUMENTS_FOR_UDF_CLASS",
+      parameters = Map("num" -> "24"),
+      sqlState = "0A000")
   }
 
   test("GROUPING_COLUMN_MISMATCH: not found the grouping column") {
     val groupingColMismatchEx = intercept[AnalysisException] {
       courseSales.cube("course", "year").agg(grouping("earnings")).explain()
     }
-    checkErrorClass(
+    checkError(
       exception = groupingColMismatchEx,
       errorClass = "GROUPING_COLUMN_MISMATCH",
-      msg =
-        "Column of grouping \\(earnings.*\\) can't be found in grouping columns course.*,year.*",
+      errorSubClass = None,
+      parameters = Map("grouping" -> "earnings.*", "groupingColumns" -> "course.*,year.*"),
       sqlState = Some("42000"),
-      matchMsg = true)
+      matchPVals = true)
   }
 
   test("GROUPING_ID_COLUMN_MISMATCH: columns of grouping_id does not match") {
     val groupingIdColMismatchEx = intercept[AnalysisException] {
       courseSales.cube("course", "year").agg(grouping_id("earnings")).explain()
     }
-    checkErrorClass(
+    checkError(
       exception = groupingIdColMismatchEx,
       errorClass = "GROUPING_ID_COLUMN_MISMATCH",
-      msg = "Columns of grouping_id \\(earnings.*\\) does not match " +
-        "grouping columns \\(course.*,year.*\\)",
+      errorSubClass = None,
+      parameters = Map("groupingIdColumn" -> "earnings.*",
+      "groupByColumns" -> "course.*,year.*"),
       sqlState = Some("42000"),
-      matchMsg = true)
+      matchPVals = true)
   }
 
   test("GROUPING_SIZE_LIMIT_EXCEEDED: max size of grouping set") {
@@ -320,17 +308,17 @@ class QueryCompilationErrorsSuite
       }
 
       withSQLConf(SQLConf.LEGACY_INTEGER_GROUPING_ID.key -> "true") {
-        checkErrorClass(
+        checkError(
           exception = intercept[AnalysisException] { testGroupingIDs(33) },
           errorClass = "GROUPING_SIZE_LIMIT_EXCEEDED",
-          msg = "Grouping sets size cannot be greater than 32")
+          parameters = Map("maxSize" -> "32"))
       }
 
       withSQLConf(SQLConf.LEGACY_INTEGER_GROUPING_ID.key -> "false") {
-        checkErrorClass(
+        checkError(
           exception = intercept[AnalysisException] { testGroupingIDs(65) },
           errorClass = "GROUPING_SIZE_LIMIT_EXCEEDED",
-          msg = "Grouping sets size cannot be greater than 64")
+          parameters = Map("maxSize" -> "64"))
       }
     }
   }
@@ -350,13 +338,13 @@ class QueryCompilationErrorsSuite
       withTempView(tempViewName) {
         sql(s"CREATE TEMPORARY VIEW $tempViewName as SELECT * FROM $tableName")
 
-        checkErrorClass(
+        checkError(
           exception = intercept[AnalysisException] {
             sql(s"DESC TABLE $tempViewName PARTITION (c='Us', d=1)")
           },
           errorClass = "FORBIDDEN_OPERATION",
-          msg = s"""The operation DESC PARTITION is not allowed """ +
-            s"on the temporary view: `$tempViewName`")
+          parameters = Map("statement" -> "DESC PARTITION",
+            "objectType" -> "TEMPORARY VIEW", "objectName" -> s"`$tempViewName`"))
       }
     }
   }
@@ -376,13 +364,13 @@ class QueryCompilationErrorsSuite
       withView(viewName) {
         sql(s"CREATE VIEW $viewName as SELECT * FROM $tableName")
 
-        checkErrorClass(
+        checkError(
           exception = intercept[AnalysisException] {
             sql(s"DESC TABLE $viewName PARTITION (c='Us', d=1)")
           },
           errorClass = "FORBIDDEN_OPERATION",
-          msg = s"""The operation DESC PARTITION is not allowed """ +
-            s"on the view: `$viewName`")
+          parameters = Map("statement" -> "DESC PARTITION",
+          "objectType" -> "VIEW", "objectName" -> s"`$viewName`"))
       }
     }
   }
@@ -390,13 +378,13 @@ class QueryCompilationErrorsSuite
   test("SECOND_FUNCTION_ARGUMENT_NOT_INTEGER: " +
     "the second argument of 'date_add' function needs to be an integer") {
     withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
-      checkErrorClass(
+      checkError(
         exception = intercept[AnalysisException] {
           sql("select date_add('1982-08-15', 'x')").collect()
         },
         errorClass = "SECOND_FUNCTION_ARGUMENT_NOT_INTEGER",
-        msg = "The second argument of 'date_add' function needs to be an integer.",
-        sqlState = Some("22023"))
+        parameters = Map("functionName" -> "date_add"),
+        sqlState = "22023")
     }
   }
 
@@ -404,13 +392,12 @@ class QueryCompilationErrorsSuite
     val schema = StructType(
       StructField("map", MapType(IntegerType, IntegerType, true), false) :: Nil)
 
-    checkErrorClass(
+    checkError(
       exception = intercept[AnalysisException] {
         spark.read.schema(schema).json(spark.emptyDataset[String])
       },
       errorClass = "INVALID_JSON_SCHEMA_MAP_TYPE",
-      msg = """Input schema "STRUCT<map: MAP<INT, INT>>" """ +
-        "can only contain STRING as a key type for a MAP."
+      parameters = Map("jsonSchema" -> "\"STRUCT<map: MAP<INT, INT>>\"")
     )
   }
 
@@ -502,7 +489,7 @@ class QueryCompilationErrorsSuite
       ("Java", 2013, 30000)
     ).toDF("course", "year", "earnings")
 
-    checkErrorClass(
+    checkError(
       exception = intercept[AnalysisException] {
         df.groupBy(df("course")).pivot(df("year"), Seq(
           struct(lit("dotnet"), lit("Experts")),
@@ -510,8 +497,9 @@ class QueryCompilationErrorsSuite
           agg(sum($"earnings")).collect()
       },
       errorClass = "PIVOT_VALUE_DATA_TYPE_MISMATCH",
-      msg = "Invalid pivot value 'struct(col1, dotnet, col2, Experts)': value data type " +
-        "struct<col1:string,col2:string> does not match pivot column data type int")
+      parameters = Map("value" -> "struct(col1, dotnet, col2, Experts)",
+        "valueType" -> "struct<col1:string,col2:string>",
+        "pivotType" -> "int"))
   }
 
   test("INVALID_FIELD_NAME: add a nested field for not struct parent") {
@@ -537,25 +525,26 @@ class QueryCompilationErrorsSuite
       ("Java", 2013, 30000)
     ).toDF("course", "year", "earnings")
 
-    checkErrorClass(
+    checkError(
       exception = intercept[AnalysisException] {
         df.groupBy(df("course")).
           pivot(df("year"), Seq($"earnings")).
           agg(sum($"earnings")).collect()
       },
       errorClass = "NON_LITERAL_PIVOT_VALUES",
-      msg = """Literal expressions required for pivot values, found "earnings".""")
+      parameters = Map("expression" -> "\"earnings\""))
   }
 
   test("UNSUPPORTED_DESERIALIZER: data type mismatch") {
     val e = intercept[AnalysisException] {
       sql("select 1 as arr").as[ArrayClass]
     }
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_DESERIALIZER",
       errorSubClass = Some("DATA_TYPE_MISMATCH"),
-      msg = """The deserializer is not supported: need an "ARRAY" field but got "INT".""")
+      parameters = Map("desiredType" -> "\"ARRAY\"", "dataType" -> "\"INT\""),
+      sqlState = None)
   }
 
   test("UNSUPPORTED_DESERIALIZER: " +
@@ -565,22 +554,24 @@ class QueryCompilationErrorsSuite
     val e1 = intercept[AnalysisException] {
       ds.as[(String, Int, Long)]
     }
-    checkErrorClass(
+    checkError(
       exception = e1,
       errorClass = "UNSUPPORTED_DESERIALIZER",
       errorSubClass = Some("FIELD_NUMBER_MISMATCH"),
-      msg = "The deserializer is not supported: try to map \"STRUCT<a: STRING, b: INT>\" " +
-        "to Tuple3, but failed as the number of fields does not line up.")
+      parameters = Map("schema" -> "\"STRUCT<a: STRING, b: INT>\"",
+        "ordinal" -> "3"),
+      sqlState = None)
 
     val e2 = intercept[AnalysisException] {
       ds.as[Tuple1[String]]
     }
-    checkErrorClass(
+    checkError(
       exception = e2,
       errorClass = "UNSUPPORTED_DESERIALIZER",
       errorSubClass = Some("FIELD_NUMBER_MISMATCH"),
-      msg = "The deserializer is not supported: try to map \"STRUCT<a: STRING, b: INT>\" " +
-        "to Tuple1, but failed as the number of fields does not line up.")
+      parameters = Map("schema" -> "\"STRUCT<a: STRING, b: INT>\"",
+        "ordinal" -> "1"),
+      sqlState = None)
   }
 
   test("UNSUPPORTED_GENERATOR: " +
@@ -589,12 +580,12 @@ class QueryCompilationErrorsSuite
       sql("""select explode(Array(1, 2, 3)) + 1""").collect()
     )
 
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_GENERATOR",
       errorSubClass = Some("NESTED_IN_EXPRESSIONS"),
-      msg = """The generator is not supported: """ +
-        """nested in expressions "(explode(array(1, 2, 3)) + 1)"""")
+      parameters = Map("expression" -> "\"(explode(array(1, 2, 3)) + 1)\""),
+      sqlState = None)
   }
 
   test("UNSUPPORTED_GENERATOR: only one generator allowed") {
@@ -602,13 +593,13 @@ class QueryCompilationErrorsSuite
       sql("""select explode(Array(1, 2, 3)), explode(Array(1, 2, 3))""").collect()
     )
 
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_GENERATOR",
       errorSubClass = Some("MULTI_GENERATOR"),
-      msg = "The generator is not supported: only one generator allowed per select clause " +
-        """but found 2: "explode(array(1, 2, 3))", "explode(array(1, 2, 3))""""
-    )
+      parameters = Map("clause" -> "SELECT", "num" -> "2",
+        "generators" -> "\"explode(array(1, 2, 3))\", \"explode(array(1, 2, 3))\""),
+      sqlState = None)
   }
 
   test("UNSUPPORTED_GENERATOR: generators are not supported outside the SELECT clause") {
@@ -616,13 +607,12 @@ class QueryCompilationErrorsSuite
       sql("""select 1 from t order by explode(Array(1, 2, 3))""").collect()
     )
 
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "UNSUPPORTED_GENERATOR",
       errorSubClass = Some("OUTSIDE_SELECT"),
-      msg = "The generator is not supported: outside the SELECT clause, found: " +
-        "'Sort [explode(array(1, 2, 3)) ASC NULLS FIRST], true"
-    )
+      parameters = Map("plan" -> "'Sort [explode(array(1, 2, 3)) ASC NULLS FIRST], true"),
+      sqlState = None)
   }
 
   test("UNSUPPORTED_GENERATOR: not a generator") {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryErrorsSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryErrorsSuiteBase.scala
index 8ae5cf2992360..895a72efeece8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryErrorsSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryErrorsSuiteBase.scala
@@ -67,6 +67,7 @@ trait QueryErrorsSuiteBase extends SharedSparkSession {
       errorClass
     }
     assert(exception.getErrorClass === errorClass)
+    assert(exception.getErrorSubClass === errorSubClass.orNull)
     assert(exception.getSqlState === sqlState)
     assert(exception.getMessage === s"""\n[$fullErrorClass] """ + message)
   }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionAnsiErrorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionAnsiErrorsSuite.scala
index 1233030162c62..368948ed8b364 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionAnsiErrorsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionAnsiErrorsSuite.scala
@@ -27,17 +27,16 @@ class QueryExecutionAnsiErrorsSuite extends QueryTest with QueryErrorsSuiteBase
   private val ansiConf = "\"" + SQLConf.ANSI_ENABLED.key + "\""
 
   test("CAST_OVERFLOW: from timestamp to int") {
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkArithmeticException] {
         sql("select CAST(TIMESTAMP '9999-12-31T12:13:14.56789Z' AS INT)").collect()
       },
       errorClass = "CAST_OVERFLOW",
-      msg =
-        "The value TIMESTAMP '9999-12-31 04:13:14.56789' of the type \"TIMESTAMP\" cannot be cast" +
-        " to \"INT\" due to an overflow. Use `try_cast` to tolerate overflow and return " +
-          "NULL instead. " +
-        s"""If necessary set $ansiConf to "false" to bypass this error.""",
-      sqlState = Some("22005"))
+      parameters = Map("value" -> "TIMESTAMP '9999-12-31 04:13:14.56789'",
+        "sourceType" -> "\"TIMESTAMP\"",
+        "targetType" -> "\"INT\"",
+        "ansiConfig" -> ansiConf),
+      sqlState = "22005")
   }
 
   test("DIVIDE_BY_ZERO: can't divide an integer by zero") {
@@ -59,14 +58,13 @@ class QueryExecutionAnsiErrorsSuite extends QueryTest with QueryErrorsSuiteBase
   }
 
   test("INVALID_FRACTION_OF_SECOND: in the function make_timestamp") {
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkDateTimeException] {
         sql("select make_timestamp(2012, 11, 30, 9, 19, 60.66666666)").collect()
       },
       errorClass = "INVALID_FRACTION_OF_SECOND",
-      msg = "The fraction of sec must be zero. Valid range is [0, 60]. " +
-        s"""If necessary set $ansiConf to "false" to bypass this error.""",
-      sqlState = Some("22023"))
+      parameters = Map("ansiConfig" -> ansiConf),
+      sqlState = "22023")
   }
 
   test("CANNOT_CHANGE_DECIMAL_PRECISION: cast string to decimal") {
@@ -87,26 +85,22 @@ class QueryExecutionAnsiErrorsSuite extends QueryTest with QueryErrorsSuiteBase
   }
 
   test("INVALID_ARRAY_INDEX: get element from array") {
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkArrayIndexOutOfBoundsException] {
         sql("select array(1, 2, 3, 4, 5)[8]").collect()
       },
       errorClass = "INVALID_ARRAY_INDEX",
-      msg = "The index 8 is out of bounds. The array has 5 elements. " +
-        s"""If necessary set $ansiConf to "false" to bypass this error."""
+      parameters = Map("indexValue" -> "8", "arraySize" -> "5", "ansiConfig" -> ansiConf)
     )
   }
 
   test("INVALID_ARRAY_INDEX_IN_ELEMENT_AT: element_at from array") {
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkArrayIndexOutOfBoundsException] {
         sql("select element_at(array(1, 2, 3, 4, 5), 8)").collect()
       },
       errorClass = "INVALID_ARRAY_INDEX_IN_ELEMENT_AT",
-      msg = "The index 8 is out of bounds. The array has 5 elements. " +
-        "Use `try_element_at` to tolerate accessing element at invalid index and return " +
-        "NULL instead. " +
-        s"""If necessary set $ansiConf to "false" to bypass this error."""
+      parameters = Map("indexValue" -> "8", "arraySize" -> "5", "ansiConfig" -> ansiConf)
     )
   }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
index 64bf1cc04f98e..ff269271ebcd1 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala
@@ -69,33 +69,35 @@ class QueryExecutionErrorsSuite
 
   test("INVALID_PARAMETER_VALUE: invalid key lengths in AES functions") {
     val (df1, df2) = getAesInputs()
-    def checkInvalidKeyLength(df: => DataFrame): Unit = {
-      checkErrorClass(
+    def checkInvalidKeyLength(df: => DataFrame, inputBytes: Int): Unit = {
+      checkError(
         exception = intercept[SparkException] {
           df.collect
         }.getCause.asInstanceOf[SparkRuntimeException],
         errorClass = "INVALID_PARAMETER_VALUE",
-        msg = "The value of parameter\\(s\\) 'key' in the `aes_encrypt`/`aes_decrypt` function " +
-          "is invalid: expects a binary value with 16, 24 or 32 bytes, but got \\d+ bytes.",
-        sqlState = Some("22023"),
-        matchMsg = true)
+        parameters = Map("parameter" -> "key",
+          "functionName" -> "`aes_encrypt`/`aes_decrypt`",
+          "expected" -> ("expects a binary value with 16, 24 or 32 bytes, but got " +
+            inputBytes.toString + " bytes.")),
+        sqlState = "22023")
     }
 
     // Encryption failure - invalid key length
-    checkInvalidKeyLength(df1.selectExpr("aes_encrypt(value, '12345678901234567')"))
-    checkInvalidKeyLength(df1.selectExpr("aes_encrypt(value, binary('123456789012345'))"))
-    checkInvalidKeyLength(df1.selectExpr("aes_encrypt(value, binary(''))"))
+    checkInvalidKeyLength(df1.selectExpr("aes_encrypt(value, '12345678901234567')"), 17)
+    checkInvalidKeyLength(df1.selectExpr("aes_encrypt(value, binary('123456789012345'))"),
+      15)
+    checkInvalidKeyLength(df1.selectExpr("aes_encrypt(value, binary(''))"), 0)
 
     // Decryption failure - invalid key length
     Seq("value16", "value24", "value32").foreach { colName =>
       checkInvalidKeyLength(df2.selectExpr(
-        s"aes_decrypt(unbase64($colName), '12345678901234567')"))
+        s"aes_decrypt(unbase64($colName), '12345678901234567')"), 17)
       checkInvalidKeyLength(df2.selectExpr(
-        s"aes_decrypt(unbase64($colName), binary('123456789012345'))"))
+        s"aes_decrypt(unbase64($colName), binary('123456789012345'))"), 15)
       checkInvalidKeyLength(df2.selectExpr(
-        s"aes_decrypt(unbase64($colName), '')"))
+        s"aes_decrypt(unbase64($colName), '')"), 0)
       checkInvalidKeyLength(df2.selectExpr(
-        s"aes_decrypt(unbase64($colName), binary(''))"))
+        s"aes_decrypt(unbase64($colName), binary(''))"), 0)
     }
   }
 
@@ -105,17 +107,17 @@ class QueryExecutionErrorsSuite
       ("value16", "1234567812345678"),
       ("value24", "123456781234567812345678"),
       ("value32", "12345678123456781234567812345678")).foreach { case (colName, key) =>
-      checkErrorClass(
+      checkError(
         exception = intercept[SparkException] {
           df2.selectExpr(s"aes_decrypt(unbase64($colName), binary('$key'), 'ECB')").collect
         }.getCause.asInstanceOf[SparkRuntimeException],
         errorClass = "INVALID_PARAMETER_VALUE",
-        msg =
-          "The value of parameter(s) 'expr, key' in the `aes_encrypt`/`aes_decrypt` function " +
-          "is invalid: Detail message: " +
-          "Given final block not properly padded. " +
-          "Such issues can arise if a bad key is used during decryption.",
-        sqlState = Some("22023"))
+        parameters = Map("parameter" -> "expr, key",
+          "functionName" -> "`aes_encrypt`/`aes_decrypt`",
+          "expected" -> ("Detail message: " +
+            "Given final block not properly padded. " +
+            "Such issues can arise if a bad key is used during decryption.")),
+        sqlState = "22023")
     }
   }
 
@@ -123,42 +125,47 @@ class QueryExecutionErrorsSuite
     val key16 = "abcdefghijklmnop"
     val key32 = "abcdefghijklmnop12345678ABCDEFGH"
     val (df1, df2) = getAesInputs()
-    def checkUnsupportedMode(df: => DataFrame): Unit = {
-      checkErrorClass(
+    def checkUnsupportedMode(df: => DataFrame, mode: String, padding: String): Unit = {
+      checkError(
         exception = intercept[SparkException] {
           df.collect
         }.getCause.asInstanceOf[SparkRuntimeException],
         errorClass = "UNSUPPORTED_FEATURE",
-        errorSubClass = Some("AES_MODE"),
-        msg =
-          """The feature is not supported: AES-\w+ with the padding \w+""" +
-          " by the `aes_encrypt`/`aes_decrypt` function.",
-        sqlState = Some("0A000"),
-        matchMsg = true)
+        errorSubClass = "AES_MODE",
+        parameters = Map("mode" -> mode,
+        "padding" -> padding,
+        "functionName" -> "`aes_encrypt`/`aes_decrypt`"),
+        sqlState = "0A000")
     }
 
     // Unsupported AES mode and padding in encrypt
-    checkUnsupportedMode(df1.selectExpr(s"aes_encrypt(value, '$key16', 'CBC')"))
-    checkUnsupportedMode(df1.selectExpr(s"aes_encrypt(value, '$key16', 'ECB', 'NoPadding')"))
+    checkUnsupportedMode(df1.selectExpr(s"aes_encrypt(value, '$key16', 'CBC')"),
+      "CBC", "DEFAULT")
+    checkUnsupportedMode(df1.selectExpr(s"aes_encrypt(value, '$key16', 'ECB', 'NoPadding')"),
+      "ECB", "NoPadding")
 
     // Unsupported AES mode and padding in decrypt
-    checkUnsupportedMode(df2.selectExpr(s"aes_decrypt(value16, '$key16', 'GSM')"))
-    checkUnsupportedMode(df2.selectExpr(s"aes_decrypt(value16, '$key16', 'GCM', 'PKCS')"))
-    checkUnsupportedMode(df2.selectExpr(s"aes_decrypt(value32, '$key32', 'ECB', 'None')"))
+    checkUnsupportedMode(df2.selectExpr(s"aes_decrypt(value16, '$key16', 'GSM')"),
+    "GSM", "DEFAULT")
+    checkUnsupportedMode(df2.selectExpr(s"aes_decrypt(value16, '$key16', 'GCM', 'PKCS')"),
+    "GCM", "PKCS")
+    checkUnsupportedMode(df2.selectExpr(s"aes_decrypt(value32, '$key32', 'ECB', 'None')"),
+    "ECB", "None")
   }
 
   test("UNSUPPORTED_FEATURE: unsupported types (map and struct) in lit()") {
-    def checkUnsupportedTypeInLiteral(v: Any): Unit = {
-      checkErrorClass(
+    def checkUnsupportedTypeInLiteral(v: Any, literal: String, dataType: String): Unit = {
+      checkError(
         exception = intercept[SparkRuntimeException] { lit(v) },
         errorClass = "UNSUPPORTED_FEATURE",
-        errorSubClass = Some("LITERAL_TYPE"),
-        msg = """The feature is not supported: Literal for '.+' of .+\.""",
-        sqlState = Some("0A000"),
-        matchMsg = true)
+        errorSubClass = "LITERAL_TYPE",
+        parameters = Map("value" -> literal, "type" -> dataType),
+        sqlState = "0A000")
     }
-    checkUnsupportedTypeInLiteral(Map("key1" -> 1, "key2" -> 2))
-    checkUnsupportedTypeInLiteral(("mike", 29, 1.0))
+    checkUnsupportedTypeInLiteral(Map("key1" -> 1, "key2" -> 2),
+      "Map(key1 -> 1, key2 -> 2)",
+      "class scala.collection.immutable.Map$Map2")
+    checkUnsupportedTypeInLiteral(("mike", 29, 1.0), "(mike,29,1.0)", "class scala.Tuple3")
 
     val e2 = intercept[SparkRuntimeException] {
       trainingSales
@@ -167,13 +174,13 @@ class QueryExecutionErrorsSuite
         .agg(sum($"sales.earnings"))
         .collect()
     }
-    checkErrorClass(
+    checkError(
       exception = e2,
       errorClass = "UNSUPPORTED_FEATURE",
-      errorSubClass = Some("PIVOT_TYPE"),
-      msg = "The feature is not supported: Pivoting by the value" +
-        """ '[dotnet,Dummies]' of the column data type "STRUCT<col1: STRING, training: STRING>".""",
-      sqlState = Some("0A000"))
+      errorSubClass = "PIVOT_TYPE",
+      parameters = Map("value" -> "[dotnet,Dummies]",
+      "type" -> "\"STRUCT<col1: STRING, training: STRING>\""),
+      sqlState = "0A000")
   }
 
   test("UNSUPPORTED_FEATURE: unsupported pivot operations") {
@@ -185,12 +192,12 @@ class QueryExecutionErrorsSuite
         .agg(sum($"sales.earnings"))
         .collect()
     }
-    checkErrorClass(
+    checkError(
       exception = e1,
       errorClass = "UNSUPPORTED_FEATURE",
-      errorSubClass = Some("REPEATED_PIVOT"),
-      msg = "The feature is not supported: Repeated PIVOT operation.",
-      sqlState = Some("0A000"))
+      errorSubClass = "REPEATED_PIVOT",
+      parameters = Map[String, String](),
+      sqlState = "0A000")
 
     val e2 = intercept[SparkUnsupportedOperationException] {
       trainingSales
@@ -199,12 +206,12 @@ class QueryExecutionErrorsSuite
         .agg(sum($"sales.earnings"))
         .collect()
     }
-    checkErrorClass(
+    checkError(
       exception = e2,
       errorClass = "UNSUPPORTED_FEATURE",
-      errorSubClass = Some("PIVOT_AFTER_GROUP_BY"),
-      msg = "The feature is not supported: PIVOT clause following a GROUP BY clause.",
-      sqlState = Some("0A000"))
+      errorSubClass = "PIVOT_AFTER_GROUP_BY",
+      parameters = Map[String, String](),
+      sqlState = "0A000")
   }
 
   test("INCONSISTENT_BEHAVIOR_CROSS_VERSION: " +
@@ -221,22 +228,14 @@ class QueryExecutionErrorsSuite
       val format = "Parquet"
       val config = "\"" + SQLConf.PARQUET_REBASE_MODE_IN_READ.key + "\""
       val option = "\"datetimeRebaseMode\""
-      checkErrorClass(
+      checkError(
         exception = e,
         errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
-        errorSubClass = Some("READ_ANCIENT_DATETIME"),
-        msg =
-          "You may get a different result due to the upgrading to Spark >= 3.0:" +
-          s"""
-            |reading dates before 1582-10-15 or timestamps before 1900-01-01T00:00:00Z
-            |from $format files can be ambiguous, as the files may be written by
-            |Spark 2.x or legacy versions of Hive, which uses a legacy hybrid calendar
-            |that is different from Spark 3.0+'s Proleptic Gregorian calendar.
-            |See more details in SPARK-31404. You can set the SQL config $config or
-            |the datasource option $option to "LEGACY" to rebase the datetime values
-            |w.r.t. the calendar difference during reading. To read the datetime values
-            |as it is, set the SQL config $config or the datasource option $option
-            |to "CORRECTED".""".stripMargin)
+        errorSubClass = "READ_ANCIENT_DATETIME",
+        parameters = Map("format" -> format,
+        "config" -> config,
+        "option" -> option),
+        sqlState = null)
     }
 
     // Fail to write ancient datetime values.
@@ -249,22 +248,13 @@ class QueryExecutionErrorsSuite
 
         val format = "Parquet"
         val config = "\"" + SQLConf.PARQUET_REBASE_MODE_IN_WRITE.key + "\""
-        checkErrorClass(
+        checkError(
           exception = e,
           errorClass = "INCONSISTENT_BEHAVIOR_CROSS_VERSION",
-          errorSubClass = Some("WRITE_ANCIENT_DATETIME"),
-          msg =
-            "You may get a different result due to the upgrading to Spark >= 3.0:" +
-            s"""
-              |writing dates before 1582-10-15 or timestamps before 1900-01-01T00:00:00Z
-              |into $format files can be dangerous, as the files may be read by Spark 2.x
-              |or legacy versions of Hive later, which uses a legacy hybrid calendar that
-              |is different from Spark 3.0+'s Proleptic Gregorian calendar. See more
-              |details in SPARK-31404. You can set $config to "LEGACY" to rebase the
-              |datetime values w.r.t. the calendar difference during writing, to get maximum
-              |interoperability. Or set $config to "CORRECTED" to write the datetime
-              |values as it is, if you are sure that the written files will only be read by
-              |Spark 3.0+ or other systems that use Proleptic Gregorian calendar.""".stripMargin)
+          errorSubClass = "WRITE_ANCIENT_DATETIME",
+          parameters = Map("format" -> format,
+            "config" -> config),
+          sqlState = null)
       }
     }
   }
@@ -273,14 +263,15 @@ class QueryExecutionErrorsSuite
     withTempPath { file =>
       sql("select timestamp_ltz'2019-03-21 00:02:03'").write.orc(file.getCanonicalPath)
       withAllNativeOrcReaders {
-        checkErrorClass(
+        checkError(
           exception = intercept[SparkException] {
             spark.read.schema("time timestamp_ntz").orc(file.getCanonicalPath).collect()
           }.getCause.asInstanceOf[SparkUnsupportedOperationException],
           errorClass = "UNSUPPORTED_FEATURE",
-          errorSubClass = Some("ORC_TYPE_CAST"),
-          msg = "The feature is not supported: " +
-            "Unable to convert \"TIMESTAMP\" of Orc to data type \"TIMESTAMP_NTZ\".")
+          errorSubClass = "ORC_TYPE_CAST",
+          parameters = Map("orcType" -> "\"TIMESTAMP\"",
+            "toType" -> "\"TIMESTAMP_NTZ\""),
+          sqlState = "0A000")
       }
     }
   }
@@ -289,27 +280,27 @@ class QueryExecutionErrorsSuite
     withTempPath { file =>
       sql("select timestamp_ntz'2019-03-21 00:02:03'").write.orc(file.getCanonicalPath)
       withAllNativeOrcReaders {
-        checkErrorClass(
+        checkError(
           exception = intercept[SparkException] {
             spark.read.schema("time timestamp_ltz").orc(file.getCanonicalPath).collect()
           }.getCause.asInstanceOf[SparkUnsupportedOperationException],
           errorClass = "UNSUPPORTED_FEATURE",
-          errorSubClass = Some("ORC_TYPE_CAST"),
-          msg = "The feature is not supported: " +
-            "Unable to convert \"TIMESTAMP_NTZ\" of Orc to data type \"TIMESTAMP\".")
+          errorSubClass = "ORC_TYPE_CAST",
+          parameters = Map("orcType" -> "\"TIMESTAMP_NTZ\"",
+            "toType" -> "\"TIMESTAMP\""),
+          sqlState = "0A000")
       }
     }
   }
 
   test("DATETIME_OVERFLOW: timestampadd() overflows its input timestamp") {
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkArithmeticException] {
         sql("select timestampadd(YEAR, 1000000, timestamp'2022-03-09 01:02:03')").collect()
       },
       errorClass = "DATETIME_OVERFLOW",
-      msg =
-        "Datetime operation overflow: add 1000000 YEAR to TIMESTAMP '2022-03-09 01:02:03'.",
-      sqlState = Some("22008"))
+      parameters = Map("operation" -> "add 1000000 YEAR to TIMESTAMP '2022-03-09 01:02:03'"),
+      sqlState = "22008")
   }
 
   test("CANNOT_PARSE_DECIMAL: unparseable decimal") {
@@ -348,11 +339,11 @@ class QueryExecutionErrorsSuite
     val e4 = e3.getCause.asInstanceOf[BadRecordException]
     assert(e4.getCause.isInstanceOf[SparkRuntimeException])
 
-    checkErrorClass(
+    checkError(
       exception = e4.getCause.asInstanceOf[SparkRuntimeException],
       errorClass = "CANNOT_PARSE_DECIMAL",
-      msg = "Cannot parse decimal",
-      sqlState = Some("42000"))
+      parameters = Map[String, String](),
+      sqlState = "42000")
   }
 
   test("WRITING_JOB_ABORTED: read of input data fails in the middle") {
@@ -373,13 +364,13 @@ class QueryExecutionErrorsSuite
         }
         val input = spark.range(15).select(failingUdf($"id").as(Symbol("i")))
           .select($"i", -$"i" as Symbol("j"))
-        checkErrorClass(
+        checkError(
           exception = intercept[SparkException] {
             input.write.format(cls.getName).option("path", path).mode("overwrite").save()
           },
           errorClass = "WRITING_JOB_ABORTED",
-          msg = "Writing job aborted",
-          sqlState = Some("40000"))
+          parameters = Map[String, String](),
+          sqlState = "40000")
         // make sure we don't have partial data.
         assert(spark.read.format(cls.getName).option("path", path).load().collect().isEmpty)
       }
@@ -387,21 +378,26 @@ class QueryExecutionErrorsSuite
   }
 
   test("FAILED_EXECUTE_UDF: execute user defined function") {
+    val luckyCharOfWord = udf { (word: String, index: Int) => {
+      word.substring(index, index + 1)
+    }}
     val e1 = intercept[SparkException] {
       val words = Seq(("Jacek", 5), ("Agata", 5), ("Sweet", 6)).toDF("word", "index")
-      val luckyCharOfWord = udf { (word: String, index: Int) => {
-        word.substring(index, index + 1)
-      }}
       words.select(luckyCharOfWord($"word", $"index")).collect()
     }
     assert(e1.getCause.isInstanceOf[SparkException])
 
-    checkErrorClass(
+    Utils.getSimpleName(luckyCharOfWord.getClass)
+
+    checkError(
       exception = e1.getCause.asInstanceOf[SparkException],
       errorClass = "FAILED_EXECUTE_UDF",
-      msg = "Failed to execute user defined function " +
-        "\\(QueryExecutionErrorsSuite\\$\\$Lambda\\$\\d+/\\w+: \\(string, int\\) => string\\)",
-      matchMsg = true)
+      errorSubClass = None,
+      parameters = Map("functionName" -> "QueryExecutionErrorsSuite\\$\\$Lambda\\$\\d+/\\w+",
+        "signature" -> "string, int",
+        "result" -> "string"),
+      sqlState = None,
+      matchPVals = true)
   }
 
   test("INCOMPARABLE_PIVOT_COLUMN: an incomparable column of the map type") {
@@ -423,12 +419,11 @@ class QueryExecutionErrorsSuite
           | )
           |""".stripMargin).collect()
     }
-    checkErrorClass(
+    checkError(
       exception = e,
       errorClass = "INCOMPARABLE_PIVOT_COLUMN",
-      msg = "Invalid pivot column `__auto_generated_subquery_name`.`map`. " +
-        "Pivot columns must be comparable.",
-      sqlState = Some("42000"))
+      parameters = Map("columnName" -> "`__auto_generated_subquery_name`.`map`"),
+      sqlState = "42000")
   }
 
   test("UNSUPPORTED_SAVE_MODE: unsupported null saveMode whether the path exists or not") {
@@ -437,11 +432,12 @@ class QueryExecutionErrorsSuite
         val saveMode: SaveMode = null
         Seq(1, 2).toDS().write.mode(saveMode).parquet(path.getAbsolutePath)
       }
-      checkErrorClass(
+      checkError(
         exception = e1,
         errorClass = "UNSUPPORTED_SAVE_MODE",
-        errorSubClass = Some("NON_EXISTENT_PATH"),
-        msg = "The save mode NULL is not supported for: a non-existent path.")
+        errorSubClass = "NON_EXISTENT_PATH",
+        parameters = Map("saveMode" -> "NULL"),
+        sqlState = null)
 
       Utils.createDirectory(path)
 
@@ -449,11 +445,12 @@ class QueryExecutionErrorsSuite
         val saveMode: SaveMode = null
         Seq(1, 2).toDS().write.mode(saveMode).parquet(path.getAbsolutePath)
       }
-      checkErrorClass(
+      checkError(
         exception = e2,
         errorClass = "UNSUPPORTED_SAVE_MODE",
-        errorSubClass = Some("EXISTENT_PATH"),
-        msg = "The save mode NULL is not supported for: an existent path.")
+        errorSubClass = "EXISTENT_PATH",
+        parameters = Map("saveMode" -> "NULL"),
+        sqlState = null)
     }
   }
 
@@ -469,12 +466,15 @@ class QueryExecutionErrorsSuite
           }
           assert(e.getCause.isInstanceOf[SparkSecurityException])
 
-          checkErrorClass(
+          checkError(
             exception = e.getCause.asInstanceOf[SparkSecurityException],
             errorClass = "RESET_PERMISSION_TO_ORIGINAL",
-            msg = "Failed to set original permission .+ " +
-              "back to the created path: .+\\. Exception: .+",
-            matchMsg = true)
+            errorSubClass = None,
+            parameters = Map("permission" -> ".+",
+              "path" -> ".+",
+              "message" -> ".+"),
+            sqlState = None,
+            matchPVals = true)
       }
     }
   }
@@ -498,12 +498,11 @@ class QueryExecutionErrorsSuite
       val e = intercept[SparkClassNotFoundException] {
         sql("CREATE TABLE student (id INT, name STRING, age INT) USING org.apache.spark.sql.fake")
       }
-      checkErrorClass(
+      checkError(
         exception = e,
         errorClass = "INCOMPATIBLE_DATASOURCE_REGISTER",
-        msg = "Detected an incompatible DataSourceRegister. Please remove the incompatible library " +
-          "from classpath or upgrade it. Error: Illegal configuration-file syntax: " +
-          "META-INF/services/org.apache.spark.sql.sources.DataSourceRegister")
+        parameters = Map("message" -> ("Illegal configuration-file syntax: " +
+          "META-INF/services/org.apache.spark.sql.sources.DataSourceRegister")))
     }
   }
 
@@ -576,12 +575,13 @@ class QueryExecutionErrorsSuite
 
     JdbcDialects.registerDialect(testH2DialectUnrecognizedSQLType)
 
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkSQLException] {
         spark.read.jdbc(urlWithUserAndPass, tableName, new Properties()).collect()
       },
       errorClass = "UNRECOGNIZED_SQL_TYPE",
-      msg = s"Unrecognized SQL type $unrecognizedColumnType")
+      parameters = Map("typeName" -> unrecognizedColumnType.toString),
+      sqlState = "42000")
 
     JdbcDialects.unregisterDialect(testH2DialectUnrecognizedSQLType)
   }
@@ -600,25 +600,27 @@ class QueryExecutionErrorsSuite
 
       val aggregated = spark.table("bucketed_table").groupBy("i").count()
 
-      checkErrorClass(
+      checkError(
         exception = intercept[SparkException] {
           aggregated.count()
         },
         errorClass = "INVALID_BUCKET_FILE",
-        msg = "Invalid bucket file: .+",
-        matchMsg = true)
+        errorSubClass = None,
+        parameters = Map("path" -> ".+"),
+        sqlState = None,
+        matchPVals = true)
     }
   }
 
   test("MULTI_VALUE_SUBQUERY_ERROR: " +
     "more than one row returned by a subquery used as an expression") {
-    checkErrorClass(
+    checkError(
       exception = intercept[SparkException] {
         sql("select (select a from (select 1 as a union all select 2 as a) t) as b").collect()
       },
       errorClass = "MULTI_VALUE_SUBQUERY_ERROR",
-      msg =
-        """more than one row returned by a subquery used as an expression: """ +
+      errorSubClass = None,
+      parameters = Map("plan" ->
           """Subquery subquery#\w+, \[id=#\w+\]
             |\+\- AdaptiveSparkPlan isFinalPlan=true
             |   \+\- == Final Plan ==
@@ -633,8 +635,9 @@ class QueryExecutionErrorsSuite
             |      :  \+\- Scan OneRowRelation\[\]
             |      \+\- Project \[\w+ AS a#\w+\]
             |         \+\- Scan OneRowRelation\[\]
-            |""".stripMargin,
-      matchMsg = true)
+            |""".stripMargin),
+      sqlState = None,
+      matchPVals = true)
   }
 }