[SPARK-46608][SQL] Restore backward compatibility of JdbcDialect.classifyException

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

@@ -1179,6 +1179,11 @@
         "message" : [
           "Check that the table <tableName> exists."
         ]
+      },
+      "UNCLASSIFIED" : {
+        "message" : [
+          "<message>"
+        ]
       }
     },
     "sqlState" : "HV000"

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala

@@ -224,7 +224,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
       val e = intercept[AnalysisException] {
         sql(s"CREATE TABLE $catalogName.new_table (i INT) TBLPROPERTIES('a'='1')")
       }
-      assert(e.getErrorClass == "FAILED_JDBC.CREATE_TABLE")
+      assert(e.getErrorClass == "FAILED_JDBC.UNCLASSIFIED")
       testCreateTableWithProperty(s"$catalogName.new_table")
     }
   }
@@ -279,7 +279,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
             sql(s"CREATE index i1 ON $catalogName.new_table (col1)")
           },
           errorClass = "INDEX_ALREADY_EXISTS",
-          parameters = Map("indexName" -> "`i1`", "tableName" -> "`new_table`")
+          parameters = Map("indexName" -> "i1", "tableName" -> "new_table")
         )
 
         sql(s"DROP index i1 ON $catalogName.new_table")
@@ -304,7 +304,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
             sql(s"DROP index i1 ON $catalogName.new_table")
           },
           errorClass = "INDEX_NOT_FOUND",
-          parameters = Map("indexName" -> "`i1`", "tableName" -> "`new_table`")
+          parameters = Map("indexName" -> "i1", "tableName" -> "new_table")
         )
       }
     }

docs/sql-error-conditions-failed-jdbc-error-class.md

@@ -77,4 +77,8 @@ Rename the table `<oldName>` to `<newName>`.
 
 Check that the table `<tableName>` exists.
 
+## UNCLASSIFIED
+
+`<message>`
+
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -1183,12 +1183,14 @@ object JdbcUtils extends Logging with SQLConfHelper {
   def classifyException[T](
       errorClass: String,
       messageParameters: Map[String, String],
-      dialect: JdbcDialect)(f: => T): T = {
+      dialect: JdbcDialect,
+      description: String)(f: => T): T = {
     try {
       f
     } catch {
       case e: SparkThrowable with Throwable => throw e
-      case e: Throwable => throw dialect.classifyException(e, errorClass, messageParameters)
+      case e: Throwable =>
+        throw dialect.classifyException(e, errorClass, messageParameters, description)
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTable.scala

@@ -69,7 +69,8 @@ case class JDBCTable(ident: Identifier, schema: StructType, jdbcOptions: JDBCOpt
           "url" -> jdbcOptions.getRedactUrl(),
           "indexName" -> toSQLId(indexName),
           "tableName" -> toSQLId(name)),
-        dialect = JdbcDialects.get(jdbcOptions.url)) {
+        dialect = JdbcDialects.get(jdbcOptions.url),
+        description = s"Failed to create index $indexName in ${name()}") {
         JdbcUtils.createIndex(
           conn, indexName, ident, columns, columnsProperties, properties, jdbcOptions)
       }
@@ -90,7 +91,8 @@ case class JDBCTable(ident: Identifier, schema: StructType, jdbcOptions: JDBCOpt
           "url" -> jdbcOptions.getRedactUrl(),
           "indexName" -> toSQLId(indexName),
           "tableName" -> toSQLId(name)),
-        dialect = JdbcDialects.get(jdbcOptions.url)) {
+        dialect = JdbcDialects.get(jdbcOptions.url),
+        description = s"Failed to drop index $indexName in ${name()}") {
         JdbcUtils.dropIndex(conn, indexName, ident, jdbcOptions)
       }
     }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala

@@ -73,7 +73,8 @@ class JDBCTableCatalog extends TableCatalog
         messageParameters = Map(
           "url" -> options.getRedactUrl(),
           "namespace" -> toSQLId(namespace.toSeq)),
-        dialect) {
+        dialect,
+        description = s"Failed get tables from: ${namespace.mkString(".")}") {
         conn.getMetaData.getTables(null, schemaPattern, "%", Array("TABLE"))
       }
       new Iterator[Identifier] {
@@ -92,7 +93,8 @@ class JDBCTableCatalog extends TableCatalog
       messageParameters = Map(
         "url" -> options.getRedactUrl(),
         "tableName" -> toSQLId(ident)),
-      dialect) {
+      dialect,
+      description = s"Failed table existence check: $ident") {
       JdbcUtils.withConnection(options)(JdbcUtils.tableExists(_, writeOptions))
     }
   }
@@ -118,7 +120,8 @@ class JDBCTableCatalog extends TableCatalog
           "url" -> options.getRedactUrl(),
           "oldName" -> toSQLId(oldIdent),
           "newName" -> toSQLId(newIdent)),
-        dialect) {
+        dialect,
+        description = s"Failed table renaming from $oldIdent to $newIdent") {
         JdbcUtils.renameTable(conn, oldIdent, newIdent, options)
       }
     }
@@ -183,7 +186,8 @@ class JDBCTableCatalog extends TableCatalog
         messageParameters = Map(
           "url" -> options.getRedactUrl(),
           "tableName" -> toSQLId(ident)),
-        dialect) {
+        dialect,
+        description = s"Failed table creation: $ident") {
         JdbcUtils.createTable(conn, getTableName(ident), schema, caseSensitive, writeOptions)
       }
     }
@@ -199,7 +203,8 @@ class JDBCTableCatalog extends TableCatalog
         messageParameters = Map(
           "url" -> options.getRedactUrl(),
           "tableName" -> toSQLId(ident)),
-        dialect) {
+        dialect,
+        description = s"Failed table altering: $ident") {
         JdbcUtils.alterTable(conn, getTableName(ident), changes, options)
       }
       loadTable(ident)
@@ -214,7 +219,8 @@ class JDBCTableCatalog extends TableCatalog
           messageParameters = Map(
             "url" -> options.getRedactUrl(),
             "namespace" -> toSQLId(namespace.toSeq)),
-          dialect) {
+          dialect,
+          description = s"Failed namespace exists: ${namespace.mkString}") {
           JdbcUtils.schemaExists(conn, options, db)
         }
       }
@@ -226,7 +232,8 @@ class JDBCTableCatalog extends TableCatalog
       JdbcUtils.classifyException(
         errorClass = "FAILED_JDBC.LIST_NAMESPACES",
         messageParameters = Map("url" -> options.getRedactUrl()),
-        dialect) {
+        dialect,
+        description = s"Failed list namespaces") {
         JdbcUtils.listSchemas(conn, options)
       }
     }
@@ -279,7 +286,8 @@ class JDBCTableCatalog extends TableCatalog
           messageParameters = Map(
             "url" -> options.getRedactUrl(),
             "namespace" -> toSQLId(db)),
-          dialect) {
+          dialect,
+          description = s"Failed create name space: $db") {
           JdbcUtils.createSchema(conn, options, db, comment)
         }
       }
@@ -303,7 +311,8 @@ class JDBCTableCatalog extends TableCatalog
                   messageParameters = Map(
                     "url" -> options.getRedactUrl(),
                     "namespace" -> toSQLId(db)),
-                  dialect) {
+                  dialect,
+                  description = s"Failed create comment on name space: $db") {
                   JdbcUtils.alterSchemaComment(conn, options, db, set.value)
                 }
               }
@@ -319,7 +328,8 @@ class JDBCTableCatalog extends TableCatalog
                   messageParameters = Map(
                     "url" -> options.getRedactUrl(),
                     "namespace" -> toSQLId(db)),
-                  dialect) {
+                  dialect,
+                  description = s"Failed remove comment on name space: $db") {
                   JdbcUtils.removeSchemaComment(conn, options, db)
                 }
               }
@@ -346,7 +356,8 @@ class JDBCTableCatalog extends TableCatalog
           messageParameters = Map(
             "url" -> options.getRedactUrl(),
             "namespace" -> toSQLId(db)),
-          dialect) {
+          dialect,
+          description = s"Failed drop name space: $db") {
           JdbcUtils.dropSchema(conn, options, db, cascade)
           true
         }

sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala

@@ -144,22 +144,16 @@ private object DB2Dialect extends JdbcDialect {
     s"COMMENT ON SCHEMA ${quoteIdentifier(schema)} IS ''"
   }
 
-  override def classifyException(
-      e: Throwable,
-      errorClass: String,
-      messageParameters: Map[String, String]): AnalysisException = {
+  override def classifyException(message: String, e: Throwable): AnalysisException = {
     e match {
       case sqlException: SQLException =>
         sqlException.getSQLState match {
           // https://www.ibm.com/docs/en/db2/11.5?topic=messages-sqlstate
-          case "42893" =>
-            throw NonEmptyNamespaceException(
-              namespace = messageParameters.get("namespace").toArray,
-              details = sqlException.getMessage,
-              cause = Some(e))
-          case _ => super.classifyException(e, errorClass, messageParameters)
+          case "42893" => throw NonEmptyNamespaceException(
+            namespace = Array.empty, details = message, cause = Some(e))
+          case _ => super.classifyException(message, e)
         }
-      case _ => super.classifyException(e, errorClass, messageParameters)
+      case _ => super.classifyException(message, e)
     }
   }
 

sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala

@@ -28,7 +28,8 @@ import scala.util.control.NonFatal
 import org.apache.commons.lang3.StringUtils
 
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException, UnresolvedAttribute}
+import org.apache.spark.sql.catalyst.util.quoteNameParts
 import org.apache.spark.sql.connector.catalog.Identifier
 import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
 import org.apache.spark.sql.connector.catalog.index.TableIndex
@@ -194,10 +195,7 @@ private[sql] object H2Dialect extends JdbcDialect {
     (ident.namespace() :+ indexName).map(quoteIdentifier).mkString(".")
   }
 
-  override def classifyException(
-      e: Throwable,
-      errorClass: String,
-      messageParameters: Map[String, String]): AnalysisException = {
+  override def classifyException(message: String, e: Throwable): AnalysisException = {
     e match {
       case exception: SQLException =>
         // Error codes are from https://www.h2database.com/javadoc/org/h2/api/ErrorCode.html
@@ -208,16 +206,15 @@ private[sql] object H2Dialect extends JdbcDialect {
             val regex = """"((?:[^"\\]|\\[\\"ntbrf])+)"""".r
             val name = regex.findFirstMatchIn(e.getMessage).get.group(1)
             val quotedName = org.apache.spark.sql.catalyst.util.quoteIdentifier(name)
-            throw new TableAlreadyExistsException(
-              errorClass = "TABLE_OR_VIEW_ALREADY_EXISTS",
+            throw new TableAlreadyExistsException(errorClass = "TABLE_OR_VIEW_ALREADY_EXISTS",
               messageParameters = Map("relationName" -> quotedName),
               cause = Some(e))
           // TABLE_OR_VIEW_NOT_FOUND_1
           case 42102 =>
-            val relationName = messageParameters.getOrElse("tableName", "")
+            val quotedName = quoteNameParts(UnresolvedAttribute.parseAttributeName(message))
             throw new NoSuchTableException(
               errorClass = "TABLE_OR_VIEW_NOT_FOUND",
-              messageParameters = Map("relationName" -> relationName),
+              messageParameters = Map("relationName" -> quotedName),
               cause = Some(e))
           // SCHEMA_NOT_FOUND_1
           case 90079 =>
@@ -227,21 +224,25 @@ private[sql] object H2Dialect extends JdbcDialect {
             throw new NoSuchNamespaceException(errorClass = "SCHEMA_NOT_FOUND",
               messageParameters = Map("schemaName" -> quotedName))
           // INDEX_ALREADY_EXISTS_1
-          case 42111 if errorClass == "FAILED_JDBC.CREATE_INDEX" =>
-            val indexName = messageParameters("indexName")
-            val tableName = messageParameters("tableName")
+          case 42111 =>
+            // The message is: Failed to create index indexName in tableName
+            val regex = "(?s)Failed to create index (.*) in (.*)".r
+            val indexName = regex.findFirstMatchIn(message).get.group(1)
+            val tableName = regex.findFirstMatchIn(message).get.group(2)
             throw new IndexAlreadyExistsException(
               indexName = indexName, tableName = tableName, cause = Some(e))
           // INDEX_NOT_FOUND_1
-          case 42112 if errorClass == "FAILED_JDBC.DROP_INDEX" =>
-            val indexName = messageParameters("indexName")
-            val tableName = messageParameters("tableName")
+          case 42112 =>
+            // The message is: Failed to drop index indexName in tableName
+            val regex = "(?s)Failed to drop index (.*) in (.*)".r
+            val indexName = regex.findFirstMatchIn(message).get.group(1)
+            val tableName = regex.findFirstMatchIn(message).get.group(2)
             throw new NoSuchIndexException(indexName, tableName, cause = Some(e))
           case _ => // do nothing
         }
       case _ => // do nothing
     }
-    super.classifyException(e, errorClass, messageParameters)
+    super.classifyException(message, e)
   }
 
   override def compileExpression(expr: Expression): Option[String] = {

sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala

@@ -633,13 +633,31 @@ abstract class JdbcDialect extends Serializable with Logging {
    * @param e The dialect specific exception.
    * @param errorClass The error class assigned in the case of an unclassified `e`
    * @param messageParameters The message parameters of `errorClass`
+   * @param description The error description
    * @return `AnalysisException` or its sub-class.
    */
   def classifyException(
       e: Throwable,
       errorClass: String,
-      messageParameters: Map[String, String]): AnalysisException = {
-    new AnalysisException(errorClass, messageParameters, cause = Some(e))
+      messageParameters: Map[String, String],
+      description: String): AnalysisException = {
+    classifyException(description, e)
+  }
+
+  /**
+   * Gets a dialect exception, classifies it and wraps it by `AnalysisException`.
+   * @param message The error message to be placed to the returned exception.
+   * @param e The dialect specific exception.
+   * @return `AnalysisException` or its sub-class.
+   */
+  @deprecated("Please override the classifyException method with an error class", "4.0.0")
+  def classifyException(message: String, e: Throwable): AnalysisException = {
+    new AnalysisException(
+      errorClass = "FAILED_JDBC.UNCLASSIFIED",
+      messageParameters = Map(
+        "url" -> "jdbc:",
+        "message" -> message),
+      cause = Some(e))
   }
 
   /**

sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala

@@ -190,21 +190,15 @@ private object MsSqlServerDialect extends JdbcDialect {
     if (limit > 0) s"TOP ($limit)" else ""
   }
 
-  override def classifyException(
-      e: Throwable,
-      errorClass: String,
-      messageParameters: Map[String, String]): AnalysisException = {
+  override def classifyException(message: String, e: Throwable): AnalysisException = {
     e match {
       case sqlException: SQLException =>
         sqlException.getErrorCode match {
-          case 3729 =>
-            throw NonEmptyNamespaceException(
-              namespace = messageParameters.get("namespace").toArray,
-              details = sqlException.getMessage,
-              cause = Some(e))
-          case _ => super.classifyException(e, errorClass, messageParameters)
+          case 3729 => throw NonEmptyNamespaceException(
+            namespace = Array.empty, details = message, cause = Some(e))
+          case _ => super.classifyException(message, e)
         }
-      case _ => super.classifyException(e, errorClass, messageParameters)
+      case _ => super.classifyException(message, e)
     }
   }