[SPARK-46393][SQL][FOLLOWUP] Classify exceptions in JDBCTableCatalog.loadTable and Fix UT

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

@@ -1255,6 +1255,11 @@
           "List namespaces."
         ]
       },
+      "LOAD_TABLE" : {
+        "message" : [
+          "Load the table <tableName>."
+        ]
+      },
       "NAMESPACE_EXISTS" : {
         "message" : [
           "Check that the namespace <namespace> exists."

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

@@ -20,9 +20,8 @@ package org.apache.spark.sql.jdbc.v2
 import org.apache.logging.log4j.Level
 
 import org.apache.spark.sql.{AnalysisException, DataFrame}
-import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException, UnresolvedAttribute}
+import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException}
 import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Sample, Sort}
-import org.apache.spark.sql.catalyst.util.quoteIdentifier
 import org.apache.spark.sql.connector.catalog.{Catalogs, Identifier, TableCatalog}
 import org.apache.spark.sql.connector.catalog.index.SupportsIndex
 import org.apache.spark.sql.connector.expressions.NullOrdering
@@ -84,6 +83,17 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
 
   def testCreateTableWithProperty(tbl: String): Unit = {}
 
+  def checkErrorFailedLoadTable(e: AnalysisException, tbl: String): Unit = {
+    checkError(
+      exception = e,
+      errorClass = "FAILED_JDBC.UNCLASSIFIED",
+      parameters = Map(
+        "url" -> "jdbc:",
+        "message" -> s"Failed to load table: $tbl"
+      )
+    )
+  }
+
   test("SPARK-33034: ALTER TABLE ... add new columns") {
     withTable(s"$catalogName.alt_table") {
       sql(s"CREATE TABLE $catalogName.alt_table (ID STRING)")
@@ -122,9 +132,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     val e = intercept[AnalysisException] {
       sql(s"ALTER TABLE $catalogName.not_existing_table ADD COLUMNS (C4 STRING)")
     }
-    checkErrorTableNotFound(e, s"`$catalogName`.`not_existing_table`",
-      ExpectedContext(s"$catalogName.not_existing_table", 12,
-        11 + s"$catalogName.not_existing_table".length))
+    checkErrorFailedLoadTable(e, "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... drop column") {
@@ -146,9 +154,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     val e = intercept[AnalysisException] {
       sql(s"ALTER TABLE $catalogName.not_existing_table DROP COLUMN C1")
     }
-    checkErrorTableNotFound(e, s"`$catalogName`.`not_existing_table`",
-      ExpectedContext(s"$catalogName.not_existing_table", 12,
-        11 + s"$catalogName.not_existing_table".length))
+    checkErrorFailedLoadTable(e, "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... update column type") {
@@ -164,9 +170,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     val e = intercept[AnalysisException] {
       sql(s"ALTER TABLE $catalogName.not_existing_table ALTER COLUMN id TYPE DOUBLE")
     }
-    checkErrorTableNotFound(e, s"`$catalogName`.`not_existing_table`",
-      ExpectedContext(s"$catalogName.not_existing_table", 12,
-        11 + s"$catalogName.not_existing_table".length))
+    checkErrorFailedLoadTable(e, "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... rename column") {
@@ -194,11 +198,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     val e = intercept[AnalysisException] {
       sql(s"ALTER TABLE $catalogName.not_existing_table RENAME COLUMN ID TO C")
     }
-    checkErrorTableNotFound(e,
-      UnresolvedAttribute.parseAttributeName(s"$catalogName.not_existing_table")
-        .map(part => quoteIdentifier(part)).mkString("."),
-      ExpectedContext(s"$catalogName.not_existing_table", 12,
-        11 + s"$catalogName.not_existing_table".length))
+    checkErrorFailedLoadTable(e, "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... update column nullability") {
@@ -209,9 +209,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     val e = intercept[AnalysisException] {
       sql(s"ALTER TABLE $catalogName.not_existing_table ALTER COLUMN ID DROP NOT NULL")
     }
-    checkErrorTableNotFound(e, s"`$catalogName`.`not_existing_table`",
-      ExpectedContext(s"$catalogName.not_existing_table", 12,
-        11 + s"$catalogName.not_existing_table".length))
+    checkErrorFailedLoadTable(e, "not_existing_table")
   }
 
   test("CREATE TABLE with table comment") {

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

@@ -131,13 +131,16 @@ class JDBCTableCatalog extends TableCatalog
     checkNamespace(ident.namespace())
     val optionsWithTableName = new JDBCOptions(
       options.parameters + (JDBCOptions.JDBC_TABLE_NAME -> getTableName(ident)))
-    try {
+    JdbcUtils.classifyException(
+      errorClass = "FAILED_JDBC.LOAD_TABLE",
+      messageParameters = Map(
+        "url" -> options.getRedactUrl(),
+        "tableName" -> toSQLId(ident)),
+      dialect,
+      description = s"Failed to load table: $ident"
+    ) {
       val schema = JDBCRDD.resolveTable(optionsWithTableName)
       JDBCTable(ident, schema, optionsWithTableName)
-    } catch {
-      case e: SQLException =>
-        logWarning("Failed to load table", e)
-        throw QueryCompilationErrors.noSuchTableError(ident)
     }
   }