From 08fdff6845f3282293708bb52a7a6bd1b9de2bf8 Mon Sep 17 00:00:00 2001
From: Wenchen Fan <wenchen@databricks.com>
Date: Thu, 6 Jun 2024 16:15:24 -0700
Subject: [PATCH] Classify exceptions in JDBCTableCatalog.loadTable

---
 .../src/main/resources/error/error-conditions.json  |  5 +++++
 .../datasources/v2/jdbc/JDBCTableCatalog.scala      | 13 ++++++++-----
 2 files changed, 13 insertions(+), 5 deletions(-)

diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 7b88300737700..36d8fe1daa379 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/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."
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala
index dbd8ee5981daa..e7a3fe0f8aa7b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalog.scala
+++ b/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)
     }
   }