[SPARK-49730][SQL] classify syntax errors for pgsql, mysql, sqlserver and h2

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

@@ -1392,6 +1392,16 @@
           "Drop the namespace <namespace>."
         ]
       },
+      "EXECUTE_QUERY" : {
+        "message" : [
+          "Execution of the query: <query>."
+        ]
+      },
+      "GET_SCHEMA" : {
+        "message" : [
+          "Schema fetching for an external query or a table: <query>."
+        ]
+      },
       "GET_TABLES" : {
         "message" : [
           "Get tables from the namespace: <namespace>."
@@ -1422,6 +1432,11 @@
           "Rename the table <oldName> to <newName>."
         ]
       },
+      "SYNTAX_ERROR" : {
+        "message" : [
+          "Compilation of the query for an external DBMS: <query>."
+        ]
+      },
       "TABLE_EXISTS" : {
         "message" : [
           "Check that the table <tableName> exists."

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

@@ -134,6 +134,27 @@ class MsSqlServerIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JD
     assert(df3.collect().length == 3)
   }
 
+  test("SPARK-49730: get_schema error classification") {
+    checkErrorMatchPVals(
+      exception = intercept[AnalysisException] {
+        val schema = StructType(
+          Seq(StructField("id", IntegerType, true)))
+
+        spark.read
+          .format("jdbc")
+          .schema(schema)
+          .option("url", jdbcUrl)
+          .option("query", "SELECT * FROM non_existent_table")
+          .load()
+      },
+      condition = "FAILED_JDBC.GET_SCHEMA",
+      parameters = Map(
+        "url" -> jdbcUrl,
+        "query" ->
+          "SELECT \\* FROM \\(SELECT \\* FROM non_existent_table\\) SPARK_GEN_SUBQ_\\d+ WHERE 1=0")
+    )
+  }
+
   test("SPARK-47994: SQLServer does not support 1 or 0 as boolean type in CASE WHEN filter") {
     val df = sql(
       s"""

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

@@ -169,6 +169,42 @@ class MySQLIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCTest
     }
   }
 
+  test("SPARK-49730: get_schema error classification") {
+    checkErrorMatchPVals(
+      exception = intercept[AnalysisException] {
+        val schema = StructType(
+          Seq(StructField("id", IntegerType, true)))
+
+        spark.read
+          .format("jdbc")
+          .schema(schema)
+          .option("url", jdbcUrl)
+          .option("query", "SELECT * FROM non_existent_table")
+          .load()
+      },
+      condition = "FAILED_JDBC.GET_SCHEMA",
+      parameters = Map(
+        "url" -> jdbcUrl,
+        "query" ->
+          "SELECT \\* FROM \\(SELECT \\* FROM non_existent_table\\) SPARK_GEN_SUBQ_\\d+ WHERE 1=0")
+    )
+  }
+
+  test("SPARK-49730: create_table error classification") {
+    val e = intercept[AnalysisException] {
+      sql(s"CREATE TABLE mysql.new_table (i INT) TBLPROPERTIES('a'='1')")
+    }
+
+    checkErrorMatchPVals(
+      exception = e,
+      condition = "FAILED_JDBC.CREATE_TABLE",
+      parameters = Map(
+        "url" -> "jdbc:.*",
+        "tableName" -> s"`new_table`"
+      )
+    )
+  }
+
   override def testDatetime(tbl: String): Unit = {
     val df1 = sql(s"SELECT name FROM $tbl WHERE " +
       "dayofyear(date1) > 100 AND dayofmonth(date1) > 10 ")

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

@@ -230,6 +230,42 @@ class PostgresIntegrationSuite extends DockerJDBCIntegrationV2Suite with V2JDBCT
     }
   }
 
+  test("SPARK-49730: get_schema error classification") {
+    checkErrorMatchPVals(
+      exception = intercept[AnalysisException] {
+        val schema = StructType(
+          Seq(StructField("id", IntegerType, true)))
+
+        spark.read
+          .format("jdbc")
+          .schema(schema)
+          .option("url", jdbcUrl)
+          .option("query", "SELECT * FROM non_existent_table")
+          .load()
+      },
+      condition = "FAILED_JDBC.GET_SCHEMA",
+      parameters = Map(
+        "url" -> jdbcUrl,
+        "query" ->
+          "SELECT \\* FROM \\(SELECT \\* FROM non_existent_table\\) SPARK_GEN_SUBQ_\\d+ WHERE 1=0")
+    )
+  }
+
+  test("SPARK-49730: create_table error classification") {
+    val e = intercept[AnalysisException] {
+      sql(s"CREATE TABLE postgresql.new_table (i INT) TBLPROPERTIES('a'='1')")
+    }
+
+    checkErrorMatchPVals(
+      exception = e,
+      condition = "FAILED_JDBC.CREATE_TABLE",
+      parameters = Map(
+        "url" -> "jdbc:.*",
+        "tableName" -> s"`new_table`"
+      )
+    )
+  }
+
   override def testDatetime(tbl: String): Unit = {
     val df1 = sql(s"SELECT name FROM $tbl WHERE " +
       "dayofyear(date1) > 100 AND dayofmonth(date1) > 10 ")

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.jdbc.v2
 
 import org.apache.logging.log4j.Level
 
+import org.apache.spark.{SparkThrowable}
 import org.apache.spark.sql.{AnalysisException, DataFrame}
 import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException}
 import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Sample, Sort}
@@ -91,7 +92,7 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
   def testCreateTableWithProperty(tbl: String): Unit = {}
 
   private def checkErrorFailedJDBC(
-      e: AnalysisException,
+      e: Throwable with SparkThrowable,
       condition: String,
       tbl: String): Unit = {
     checkErrorMatchPVals(
@@ -249,16 +250,6 @@ private[v2] trait V2JDBCTest extends SharedSparkSession with DockerIntegrationFu
     }
   }
 
-  test("CREATE TABLE with table property") {
-    withTable(s"$catalogName.new_table") {
-      val e = intercept[AnalysisException] {
-        sql(s"CREATE TABLE $catalogName.new_table (i INT) TBLPROPERTIES('a'='1')")
-      }
-      checkErrorFailedJDBC(e, "FAILED_JDBC.CREATE_TABLE", "new_table")
-      testCreateTableWithProperty(s"$catalogName.new_table")
-    }
-  }
-
   def supportsIndex: Boolean = false
 
   def supportListIndexes: Boolean = false

core/src/test/scala/org/apache/spark/SparkFunSuite.scala

@@ -352,7 +352,7 @@ abstract class SparkFunSuite
         exp => {
           val parm = parameters.getOrElse(exp._1,
             throw new IllegalArgumentException("Missing parameter" + exp._1))
-          if (!exp._2.matches(parm)) {
+          if (!exp._2.matches(parm) && exp._2 != parm) {
             throw new IllegalArgumentException("For parameter '" + exp._1 + "' value '" + exp._2 +
               "' does not match: " + parm)
           }

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -1661,6 +1661,26 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
         "operation" -> operation))
   }
 
+  def jdbcGeneratedQuerySyntaxError(url: String, query: String): Throwable = {
+    new AnalysisException(
+      errorClass = "FAILED_JDBC.SYNTAX_ERROR",
+      messageParameters = Map(
+        "query" -> query,
+        "url" -> url
+      )
+    )
+  }
+
+  def jdbcGeneratedQueryGetSchemaError(url: String, query: String): Throwable = {
+    new AnalysisException(
+      errorClass = "FAILED_JDBC.GET_SCHEMA",
+      messageParameters = Map(
+        "query" -> query,
+        "url" -> url
+      )
+    )
+  }
+
   def schemaNotSpecifiedForSchemaRelationProviderError(className: String): Throwable = {
     new AnalysisException(
       errorClass = "_LEGACY_ERROR_TEMP_1132",

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -921,6 +921,16 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
         "jdbcQueryString" -> jdbcQueryString))
   }
 
+  def jdbcGeneratedQueryExecutionError(url: String, query: String): Throwable = {
+    new SparkRuntimeException(
+      errorClass = "FAILED_JDBC.EXECUTE_QUERY",
+      messageParameters = Map(
+        "query" -> query,
+        "url" -> url
+      )
+    )
+  }
+
   def missingJdbcTableNameAndQueryError(
       jdbcTableName: String, jdbcQueryString: String): SparkIllegalArgumentException = {
     new SparkIllegalArgumentException(

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

@@ -240,7 +240,18 @@ private[sql] object JDBCRelation extends Logging {
    * @return resolved Catalyst schema of a JDBC table
    */
   def getSchema(resolver: Resolver, jdbcOptions: JDBCOptions): StructType = {
-    val tableSchema = JDBCRDD.resolveTable(jdbcOptions)
+    val dialect = JdbcDialects.get(jdbcOptions.url)
+    val tableSchema = JdbcUtils.classifyException(
+      errorClass = "...",
+      messageParameters = Map(
+        "query" -> dialect.getSchemaQuery(jdbcOptions.tableOrQuery),
+        "url" -> jdbcOptions.url),
+      dialect = dialect,
+      description =
+        s"Failed to fetch schema for: ${dialect.getSchemaQuery(jdbcOptions.tableOrQuery)}",
+      isRuntime = false) {
+      JDBCRDD.resolveTable(jdbcOptions)
+    }
     jdbcOptions.customSchema match {
       case Some(customSchema) => JdbcUtils.getCustomSchema(
         tableSchema, customSchema, resolver)

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

@@ -152,6 +152,7 @@ private case class DB2Dialect() extends JdbcDialect with SQLConfHelper with NoLe
   override def removeSchemaCommentQuery(schema: String): String = {
     s"COMMENT ON SCHEMA ${quoteIdentifier(schema)} IS ''"
   }
+
   override def classifyException(
       e: Throwable,
       errorClass: String,

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

@@ -33,6 +33,7 @@ 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
 import org.apache.spark.sql.connector.expressions.{Expression, FieldReference, NamedReference}
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
 import org.apache.spark.sql.types.{BooleanType, ByteType, DataType, DecimalType, MetadataBuilder, ShortType, StringType, TimestampType}
 
@@ -240,6 +241,11 @@ private[sql] case class H2Dialect() extends JdbcDialect with NoLegacyJDBCError {
             val indexName = messageParameters("indexName")
             val tableName = messageParameters("tableName")
             throw new NoSuchIndexException(indexName, tableName, cause = Some(e))
+          // SYNTAX_ERROR_1, SYNTAX_ERROR_2
+          case 42000 | 42001 =>
+            throw QueryCompilationErrors.jdbcGeneratedQuerySyntaxError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
           case _ => // do nothing
         }
       case _ => // do nothing

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

@@ -217,9 +217,17 @@ private case class MsSqlServerDialect() extends JdbcDialect with NoLegacyJDBCErr
               namespace = messageParameters.get("namespace").toArray,
               details = sqlException.getMessage,
               cause = Some(e))
-           case 15335 if errorClass == "FAILED_JDBC.RENAME_TABLE" =>
-             val newTable = messageParameters("newName")
-             throw QueryCompilationErrors.tableAlreadyExistsError(newTable)
+          case 15335 if errorClass == "FAILED_JDBC.RENAME_TABLE" =>
+            val newTable = messageParameters("newName")
+            throw QueryCompilationErrors.tableAlreadyExistsError(newTable)
+          case 102 | 122 | 142 | 148 | 156 | 319 | 336 =>
+            throw QueryCompilationErrors.jdbcGeneratedQuerySyntaxError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
+          case 208 if errorClass == "..." => // Error code for "Invalid object name"
+            throw QueryCompilationErrors.jdbcGeneratedQueryGetSchemaError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
           case _ =>
             super.classifyException(e, errorClass, messageParameters, description, isRuntime)
         }

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

@@ -369,6 +369,14 @@ private case class MySQLDialect() extends JdbcDialect with SQLConfHelper with No
             val indexName = messageParameters("indexName")
             val tableName = messageParameters("tableName")
             throw new NoSuchIndexException(indexName, tableName, cause = Some(e))
+          case 1064 if errorClass == "..." =>
+            throw QueryCompilationErrors.jdbcGeneratedQuerySyntaxError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
+          case 1146 if errorClass == "..." =>
+            throw QueryCompilationErrors.jdbcGeneratedQueryGetSchemaError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
           case _ =>
             super.classifyException(e, errorClass, messageParameters, description, isRuntime)
         }

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

@@ -267,6 +267,14 @@ private case class PostgresDialect()
       case sqlException: SQLException =>
         sqlException.getSQLState match {
           // https://www.postgresql.org/docs/14/errcodes-appendix.html
+          case "42601" if errorClass == "..." =>
+            throw QueryCompilationErrors.jdbcGeneratedQuerySyntaxError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
+          case "42P01" if errorClass == "..." =>
+            throw QueryCompilationErrors.jdbcGeneratedQueryGetSchemaError(
+              messageParameters.get("url").getOrElse(""),
+              messageParameters.get("query").getOrElse(""))
           case "42P07" =>
             if (errorClass == "FAILED_JDBC.CREATE_INDEX") {
               throw new IndexAlreadyExistsException(

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

@@ -29,7 +29,7 @@ import scala.util.Random
 import org.mockito.ArgumentMatchers._
 import org.mockito.Mockito._
 
-import org.apache.spark.{SparkException, SparkSQLException}
+import org.apache.spark.{SparkException, SparkRuntimeException, SparkSQLException}
 import org.apache.spark.sql.{AnalysisException, DataFrame, Observation, QueryTest, Row}
 import org.apache.spark.sql.catalyst.{analysis, TableIdentifier}
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
@@ -1515,6 +1515,27 @@ class JDBCSuite extends QueryTest with SharedSparkSession {
     assert(res === (foobarCnt, 0L, foobarCnt) :: Nil)
   }
 
+  test("SPARK-49730: syntax error classification") {
+    checkErrorMatchPVals(
+      exception = intercept[SparkRuntimeException] {
+        val schema = StructType(
+          Seq(StructField("id", IntegerType, true, defaultMetadata(IntegerType))))
+
+        spark.read
+          .format("jdbc")
+          .schema(schema)
+          .option("url", urlWithUserAndPass)
+          .option("query", "SELECT * FRM tbl")
+          .load()
+      },
+      condition = "FAILED_JDBC.SYNTAX_ERROR",
+      parameters = Map(
+        "url" -> urlWithUserAndPass,
+        "query" ->
+          "SELECT \\* FROM \\(SELECT \\* FRM tbl\\) SPARK_GEN_SUBQ_\\d+ WHERE 1=0")
+    )
+  }
+
   test("unsupported types") {
     checkError(
       exception = intercept[SparkSQLException] {