[SPARK-48585][SQL] Make built-in JdbcDialect's method classifyException throw out the original exception

panbingkun · LuciferYang · commit a3feffdff9cd · 2024-06-18T14:50:34.000+08:00
### What changes were proposed in this pull request? The pr aims to make `built-in` JdbcDialect's method classifyException throw out the `original` exception. ### Why are the changes needed? As discussed in #46912 (comment), the following code: https://github.com/apache/spark/blob/df4156aa3217cf0f58b4c6cbf33c967bb43f7155/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala#L746-L751 have lost the original cause of the error, let's correct it. ### Does this PR introduce _any_ user-facing change? Yes, more accurate error conditions for end users. ### How was this patch tested? - Manually test. - Update existed UT & Pass GA. ### Was this patch authored or co-authored using generative AI tooling? No. Closes #46937 from panbingkun/improve_JDBCTableCatalog. Authored-by: panbingkun <panbingkun@baidu.com> Signed-off-by: yangjie01 <yangjie01@baidu.com>
diff --git a/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala b/connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/v2/V2JDBCTest.scala
@@ -83,14 +83,16 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
 
   def testCreateTableWithProperty(tbl: String): Unit = {}
 
-  def checkErrorFailedLoadTable(e: AnalysisException, tbl: String): Unit = {
-    checkError(
+  private def checkErrorFailedJDBC(
+      e: AnalysisException,
+      errorClass: String,
+      tbl: String): Unit = {
+    checkErrorMatchPVals(
       exception = e,
-      errorClass = "FAILED_JDBC.UNCLASSIFIED",
+      errorClass = errorClass,
       parameters = Map(
-        "url" -> "jdbc:",
-        "message" -> s"Failed to load table: $tbl"
-      )
+        "url" -> "jdbc:.*",
+        "tableName" -> s"`$tbl`")
     )
   }
 
@@ -132,7 +134,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)")
     }
-    checkErrorFailedLoadTable(e, "not_existing_table")
+    checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... drop column") {
@@ -154,7 +156,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     val e = intercept[AnalysisException] {
       sql(s"ALTER TABLE $catalogName.not_existing_table DROP COLUMN C1")
     }
-    checkErrorFailedLoadTable(e, "not_existing_table")
+    checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... update column type") {
@@ -170,7 +172,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")
     }
-    checkErrorFailedLoadTable(e, "not_existing_table")
+    checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... rename column") {
@@ -198,7 +200,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")
     }
-    checkErrorFailedLoadTable(e, "not_existing_table")
+    checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")
   }
 
   test("SPARK-33034: ALTER TABLE ... update column nullability") {
@@ -209,7 +211,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")
     }
-    checkErrorFailedLoadTable(e, "not_existing_table")
+    checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")
   }
 
   test("CREATE TABLE with table comment") {
@@ -231,7 +233,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.UNCLASSIFIED")
+      checkErrorFailedJDBC(e, "FAILED_JDBC.CREATE_TABLE", "new_table")
       testCreateTableWithProperty(s"$catalogName.new_table")
     }
   }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala
@@ -26,7 +26,8 @@ import org.apache.spark.sql.types.{DataType, MetadataBuilder}
  *
  * @param dialects List of dialects.
  */
-private class AggregatedDialect(dialects: List[JdbcDialect]) extends JdbcDialect {
+private class AggregatedDialect(dialects: List[JdbcDialect])
+  extends JdbcDialect with NoLegacyJDBCError {
 
   require(dialects.nonEmpty)
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.connector.expressions.Expression
 import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
 import org.apache.spark.sql.types._
 
-private case class DB2Dialect() extends JdbcDialect with SQLConfHelper {
+private case class DB2Dialect() extends JdbcDialect with SQLConfHelper with NoLegacyJDBCError {
 
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:db2")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DatabricksDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DatabricksDialect.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions
 import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
 import org.apache.spark.sql.types._
 
-private case class DatabricksDialect() extends JdbcDialect {
+private case class DatabricksDialect() extends JdbcDialect with NoLegacyJDBCError {
 
   override def canHandle(url: String): Boolean = {
     url.startsWith("jdbc:databricks")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala
@@ -25,7 +25,7 @@ import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors
 import org.apache.spark.sql.types._
 
 
-private case class DerbyDialect() extends JdbcDialect {
+private case class DerbyDialect() extends JdbcDialect with NoLegacyJDBCError {
 
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:derby")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala
@@ -37,7 +37,7 @@ import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, N
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
 import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DecimalType, MetadataBuilder, ShortType, StringType, TimestampType}
 
-private[sql] case class H2Dialect() extends JdbcDialect {
+private[sql] case class H2Dialect() extends JdbcDialect with NoLegacyJDBCError {
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:h2")
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala
@@ -841,6 +841,23 @@ abstract class JdbcDialect extends Serializable with Logging {
       metadata: MetadataBuilder): Unit = {}
 }
 
+/**
+ * Make the `classifyException` method throw out the original exception
+ */
+trait NoLegacyJDBCError extends JdbcDialect {
+
+  override def classifyException(
+      e: Throwable,
+      errorClass: String,
+      messageParameters: Map[String, String],
+      description: String): AnalysisException = {
+    new AnalysisException(
+      errorClass = errorClass,
+      messageParameters = messageParameters,
+      cause = Some(e))
+  }
+}
+
 /**
  * :: DeveloperApi ::
  * Registry of dialects that apply to every new jdbc `org.apache.spark.sql.DataFrame`.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.jdbc.MsSqlServerDialect.{GEOGRAPHY, GEOMETRY}
 import org.apache.spark.sql.types._
 
 
-private case class MsSqlServerDialect() extends JdbcDialect {
+private case class MsSqlServerDialect() extends JdbcDialect with NoLegacyJDBCError {
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:sqlserver")
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala
@@ -35,7 +35,7 @@ import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
 import org.apache.spark.sql.types._
 
-private case class MySQLDialect() extends JdbcDialect with SQLConfHelper {
+private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with NoLegacyJDBCError {
 
   override def canHandle(url : String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:mysql")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala
@@ -30,7 +30,7 @@ import org.apache.spark.sql.jdbc.OracleDialect._
 import org.apache.spark.sql.types._
 
 
-private case class OracleDialect() extends JdbcDialect with SQLConfHelper {
+private case class OracleDialect() extends JdbcDialect with SQLConfHelper with NoLegacyJDBCError {
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:oracle")
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala
@@ -37,7 +37,8 @@ import org.apache.spark.sql.execution.datasources.v2.TableSampleInfo
 import org.apache.spark.sql.types._
 
 
-private case class PostgresDialect() extends JdbcDialect with SQLConfHelper {
+private case class PostgresDialect()
+  extends JdbcDialect with SQLConfHelper with NoLegacyJDBCError {
 
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:postgresql")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/SnowflakeDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/SnowflakeDialect.scala
@@ -22,7 +22,7 @@ import java.util.Locale
 import org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils
 import org.apache.spark.sql.types.{BooleanType, DataType}
 
-private case class SnowflakeDialect() extends JdbcDialect {
+private case class SnowflakeDialect() extends JdbcDialect with NoLegacyJDBCError {
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:snowflake")
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala b/sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala
@@ -24,7 +24,7 @@ import org.apache.spark.sql.connector.catalog.Identifier
 import org.apache.spark.sql.types._
 
 
-private case class TeradataDialect() extends JdbcDialect {
+private case class TeradataDialect() extends JdbcDialect with NoLegacyJDBCError {
 
   override def canHandle(url: String): Boolean =
     url.toLowerCase(Locale.ROOT).startsWith("jdbc:teradata")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala
@@ -619,15 +619,15 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
 
   test("CREATE TABLE with table property") {
     withTable("h2.test.new_table") {
-      checkError(
+      checkErrorMatchPVals(
         exception = intercept[AnalysisException] {
           sql("CREATE TABLE h2.test.new_table(i INT, j STRING)" +
             " TBLPROPERTIES('ENGINE'='tableEngineName')")
         },
-        errorClass = "FAILED_JDBC.UNCLASSIFIED",
+        errorClass = "FAILED_JDBC.CREATE_TABLE",
         parameters = Map(
-          "url" -> "jdbc:",
-          "message" -> "Failed table creation: test.new_table"))
+          "url" -> "jdbc:.*",
+          "tableName" -> "`test`.`new_table`"))
     }
   }
 
@@ -639,14 +639,14 @@ class JDBCTableCatalogSuite extends QueryTest with SharedSparkSession {
   }
 
   test("SPARK-42904: CREATE TABLE with char/varchar with invalid char length") {
-    checkError(
+    checkErrorMatchPVals(
       exception = intercept[AnalysisException]{
         sql("CREATE TABLE h2.test.new_table(c CHAR(1000000001))")
       },
-      errorClass = "FAILED_JDBC.UNCLASSIFIED",
+      errorClass = "FAILED_JDBC.CREATE_TABLE",
       parameters = Map(
-        "url" -> "jdbc:",
-        "message" -> "Failed table creation: test.new_table"))
+        "url" -> "jdbc:.*",
+        "tableName" -> "`test`.`new_table`"))
   }
 
   test("SPARK-42955: Skip classifyException and wrap AnalysisException for SparkThrowable") {

Original file line number	Diff line number	Diff line change
`@@ -83,14 +83,16 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`83`	`83`
`84`	`84`	`def testCreateTableWithProperty(tbl: String): Unit = {}`
`85`	`85`
`86`		`- def checkErrorFailedLoadTable(e: AnalysisException, tbl: String): Unit = {`
`87`		`- checkError(`
	`86`	`+ private def checkErrorFailedJDBC(`
	`87`	`+ e: AnalysisException,`
	`88`	`+ errorClass: String,`
	`89`	`+ tbl: String): Unit = {`
	`90`	`+ checkErrorMatchPVals(`
`88`	`91`	`exception = e,`
`89`		`- errorClass = "FAILED_JDBC.UNCLASSIFIED",`
	`92`	`+ errorClass = errorClass,`
`90`	`93`	`parameters = Map(`
`91`		`- "url" -> "jdbc:",`
`92`		`- "message" -> s"Failed to load table: $tbl"`
`93`		`- )`
	`94`	`+ "url" -> "jdbc:.*",`
	`95`	+ "tableName" -> s"`$tbl`")
`94`	`96`	`)`
`95`	`97`	`}`
`96`	`98`
`@@ -132,7 +134,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`132`	`134`	`val e = intercept[AnalysisException] {`
`133`	`135`	`sql(s"ALTER TABLE $catalogName.not_existing_table ADD COLUMNS (C4 STRING)")`
`134`	`136`	`}`
`135`		`- checkErrorFailedLoadTable(e, "not_existing_table")`
	`137`	`+ checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")`
`136`	`138`	`}`
`137`	`139`
`138`	`140`	`test("SPARK-33034: ALTER TABLE ... drop column") {`
`@@ -154,7 +156,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`154`	`156`	`val e = intercept[AnalysisException] {`
`155`	`157`	`sql(s"ALTER TABLE $catalogName.not_existing_table DROP COLUMN C1")`
`156`	`158`	`}`
`157`		`- checkErrorFailedLoadTable(e, "not_existing_table")`
	`159`	`+ checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")`
`158`	`160`	`}`
`159`	`161`
`160`	`162`	`test("SPARK-33034: ALTER TABLE ... update column type") {`
`@@ -170,7 +172,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`170`	`172`	`val e = intercept[AnalysisException] {`
`171`	`173`	`sql(s"ALTER TABLE $catalogName.not_existing_table ALTER COLUMN id TYPE DOUBLE")`
`172`	`174`	`}`
`173`		`- checkErrorFailedLoadTable(e, "not_existing_table")`
	`175`	`+ checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")`
`174`	`176`	`}`
`175`	`177`
`176`	`178`	`test("SPARK-33034: ALTER TABLE ... rename column") {`
`@@ -198,7 +200,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`198`	`200`	`val e = intercept[AnalysisException] {`
`199`	`201`	`sql(s"ALTER TABLE $catalogName.not_existing_table RENAME COLUMN ID TO C")`
`200`	`202`	`}`
`201`		`- checkErrorFailedLoadTable(e, "not_existing_table")`
	`203`	`+ checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")`
`202`	`204`	`}`
`203`	`205`
`204`	`206`	`test("SPARK-33034: ALTER TABLE ... update column nullability") {`
`@@ -209,7 +211,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`209`	`211`	`val e = intercept[AnalysisException] {`
`210`	`212`	`sql(s"ALTER TABLE $catalogName.not_existing_table ALTER COLUMN ID DROP NOT NULL")`
`211`	`213`	`}`
`212`		`- checkErrorFailedLoadTable(e, "not_existing_table")`
	`214`	`+ checkErrorFailedJDBC(e, "FAILED_JDBC.LOAD_TABLE", "not_existing_table")`
`213`	`215`	`}`
`214`	`216`
`215`	`217`	`test("CREATE TABLE with table comment") {`
`@@ -231,7 +233,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu`
`231`	`233`	`val e = intercept[AnalysisException] {`
`232`	`234`	`sql(s"CREATE TABLE $catalogName.new_table (i INT) TBLPROPERTIES('a'='1')")`
`233`	`235`	`}`
`234`		`- assert(e.getErrorClass == "FAILED_JDBC.UNCLASSIFIED")`
	`236`	`+ checkErrorFailedJDBC(e, "FAILED_JDBC.CREATE_TABLE", "new_table")`
`235`	`237`	`testCreateTableWithProperty(s"$catalogName.new_table")`
`236`	`238`	`}`
`237`	`239`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,8 @@ import org.apache.spark.sql.types.{DataType, MetadataBuilder}`
`26`	`26`	`*`
`27`	`27`	`* @param dialects List of dialects.`
`28`	`28`	`*/`
`29`		`-private class AggregatedDialect(dialects: List[JdbcDialect]) extends JdbcDialect {`
	`29`	`+private class AggregatedDialect(dialects: List[JdbcDialect])`
	`30`	`+ extends JdbcDialect with NoLegacyJDBCError {`
`30`	`31`
`31`	`32`	`require(dialects.nonEmpty)`
`32`	`33`