[SPARK-38949][SQL] Wrap SQL statements by double quotes in error mess…

…ages

### What changes were proposed in this pull request?
In the PR, I propose to wrap any SQL statement in error messages by double quotes "", and apply new implementation of `QueryErrorsBase.toSQLStmt()` to all exceptions in `Query.*Errors` w/ error classes. Also this PR modifies all affected tests, see the list in the section "How was this patch tested?".

### Why are the changes needed?
To improve user experience with Spark SQL by highlighting SQL statements in error massage and make them more visible to users.

### Does this PR introduce _any_ user-facing change?
Yes. The changes might influence on error messages that are visible to users.

Before:
```sql
The operation DESC PARTITION is not allowed
```

After:
```sql
The operation "DESC PARTITION" is not allowed
```

### How was this patch tested?
By running affected test suites:
```
$ build/sbt "sql/testOnly *QueryExecutionErrorsSuite"
$ build/sbt "sql/testOnly *QueryParsingErrorsSuite"
$ build/sbt "sql/testOnly *QueryCompilationErrorsSuite"
$ build/sbt "test:testOnly *QueryCompilationErrorsDSv2Suite"
$ build/sbt "test:testOnly *ExtractPythonUDFFromJoinConditionSuite"
$ build/sbt "testOnly *PlanParserSuite"
$ build/sbt "sql/testOnly *SQLQueryTestSuite -- -z transform.sql"
$ build/sbt "sql/testOnly *SQLQueryTestSuite -- -z join-lateral.sql"
$ build/sbt "sql/testOnly *SQLQueryTestSuite -- -z describe.sql"
```

Closes #36259 from MaxGekk/error-class-apply-toSQLStmt.

Authored-by: Max Gekk <max.gekk@gmail.com>
Signed-off-by: Max Gekk <max.gekk@gmail.com>

python/pyspark/sql/tests/test_udf.py

@@ -258,15 +258,16 @@ def test_udf_not_supported_in_join_condition(self):
         def runWithJoinType(join_type, type_string):
             with self.assertRaisesRegex(
                 AnalysisException,
-                "Using PythonUDF in join condition of join type %s is not supported" % type_string,
+                """Using PythonUDF in join condition of join type "%s" is not supported"""
+                % type_string,
             ):
                 left.join(right, [f("a", "b"), left.a1 == right.b1], join_type).collect()
 
-        runWithJoinType("full", "FullOuter")
-        runWithJoinType("left", "LeftOuter")
-        runWithJoinType("right", "RightOuter")
-        runWithJoinType("leftanti", "LeftAnti")
-        runWithJoinType("leftsemi", "LeftSemi")
+        runWithJoinType("full", "FULL OUTER")
+        runWithJoinType("left", "LEFT OUTER")
+        runWithJoinType("right", "RIGHT OUTER")
+        runWithJoinType("leftanti", "LEFT ANTI")
+        runWithJoinType("leftsemi", "LEFT SEMI")
 
     def test_udf_as_join_condition(self):
         left = self.spark.createDataFrame([Row(a=1, a1=1, a2=1), Row(a=2, a1=2, a2=2)])

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -1166,7 +1166,7 @@ class AstBuilder extends SqlBaseParserBaseVisitor[AnyRef] with SQLConfHelper wit
         }
         if (join.LATERAL != null) {
           if (!Seq(Inner, Cross, LeftOuter).contains(joinType)) {
-            throw QueryParsingErrors.unsupportedLateralJoinTypeError(ctx, joinType.toString)
+            throw QueryParsingErrors.unsupportedLateralJoinTypeError(ctx, joinType.sql)
           }
           LateralJoin(left, LateralSubquery(plan(join.right)), joinType, condition)
         } else {

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

@@ -95,7 +95,8 @@ object QueryCompilationErrors extends QueryErrorsBase {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE",
       messageParameters = Array(
-        s"IF NOT EXISTS for the table ${toSQLId(tableName)} by INSERT INTO."))
+        s"${toSQLStmt("IF NOT EXISTS")} for the table ${toSQLId(tableName)} " +
+        s"by ${toSQLStmt("INSERT INTO")}."))
   }
 
   def nonPartitionColError(partitionName: String): Throwable = {
@@ -1573,7 +1574,8 @@ object QueryCompilationErrors extends QueryErrorsBase {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE",
       messageParameters = Array(
-        s"Using PythonUDF in join condition of join type $joinType is not supported"))
+        "Using PythonUDF in join condition of join type " +
+        s"${toSQLStmt(joinType.sql)} is not supported."))
   }
 
   def conflictingAttributesInJoinConditionError(

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

@@ -17,6 +17,8 @@
 
 package org.apache.spark.sql.errors
 
+import java.util.Locale
+
 import org.apache.spark.sql.catalyst.expressions.Literal
 import org.apache.spark.sql.catalyst.util.quoteIdentifier
 import org.apache.spark.sql.types.{DataType, DoubleType, FloatType}
@@ -48,7 +50,7 @@ trait QueryErrorsBase {
 
   // Quote sql statements in error messages.
   def toSQLStmt(text: String): String = {
-    s"'$text'"
+    "\"" + text.toUpperCase(Locale.ROOT) + "\""
   }
 
   def toSQLId(parts: Seq[String]): String = {

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

@@ -1892,13 +1892,13 @@ object QueryExecutionErrors extends QueryErrorsBase {
   def repeatedPivotsUnsupportedError(): Throwable = {
     new SparkUnsupportedOperationException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("Repeated pivots."))
+      messageParameters = Array(s"Repeated ${toSQLStmt("pivot")}s."))
   }
 
   def pivotNotAfterGroupByUnsupportedError(): Throwable = {
     new SparkUnsupportedOperationException(
       errorClass = "UNSUPPORTED_FEATURE",
-      messageParameters = Array("Pivot not after a groupBy."))
+      messageParameters = Array(s"${toSQLStmt("pivot")} not after a ${toSQLStmt("group by")}."))
   }
 
   private val aesFuncName = toSQLId("aes_encrypt") + "/" + toSQLId("aes_decrypt")

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

@@ -90,33 +90,51 @@ object QueryParsingErrors extends QueryErrorsBase {
   }
 
   def transformNotSupportQuantifierError(ctx: ParserRuleContext): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE",
-      Array("TRANSFORM does not support DISTINCT/ALL in inputs"), ctx)
+    new ParseException(
+      errorClass = "UNSUPPORTED_FEATURE",
+      messageParameters = Array(s"${toSQLStmt("TRANSFORM")} does not support" +
+        s" ${toSQLStmt("DISTINCT")}/${toSQLStmt("ALL")} in inputs"),
+      ctx)
   }
 
   def transformWithSerdeUnsupportedError(ctx: ParserRuleContext): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE",
-      Array("TRANSFORM with serde is only supported in hive mode"), ctx)
+    new ParseException(
+      errorClass = "UNSUPPORTED_FEATURE",
+      messageParameters = Array(
+        s"${toSQLStmt("TRANSFORM")} with serde is only supported in hive mode"),
+      ctx)
   }
 
   def lateralWithPivotInFromClauseNotAllowedError(ctx: FromClauseContext): Throwable = {
     new ParseException("LATERAL cannot be used together with PIVOT in FROM clause", ctx)
   }
 
   def lateralJoinWithNaturalJoinUnsupportedError(ctx: ParserRuleContext): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE", Array("LATERAL join with NATURAL join."), ctx)
+    new ParseException(
+      errorClass = "UNSUPPORTED_FEATURE",
+      messageParameters = Array(s"${toSQLStmt("LATERAL")} join with ${toSQLStmt("NATURAL")} join."),
+      ctx)
   }
 
   def lateralJoinWithUsingJoinUnsupportedError(ctx: ParserRuleContext): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE", Array("LATERAL join with USING join."), ctx)
+    new ParseException(
+      errorClass = "UNSUPPORTED_FEATURE",
+      messageParameters = Array(s"${toSQLStmt("LATERAL")} join with ${toSQLStmt("USING")} join."),
+      ctx)
   }
 
   def unsupportedLateralJoinTypeError(ctx: ParserRuleContext, joinType: String): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE", Array(s"LATERAL join type '$joinType'."), ctx)
+    new ParseException(
+      errorClass = "UNSUPPORTED_FEATURE",
+      messageParameters = Array(s"${toSQLStmt("LATERAL")} join type ${toSQLStmt(joinType)}."),
+      ctx)
   }
 
   def invalidLateralJoinRelationError(ctx: RelationPrimaryContext): Throwable = {
-    new ParseException("INVALID_SQL_SYNTAX", Array("LATERAL can only be used with subquery."), ctx)
+    new ParseException(
+      errorClass = "INVALID_SQL_SYNTAX",
+      messageParameters = Array(s"${toSQLStmt("LATERAL")} can only be used with subquery."),
+      ctx)
   }
 
   def repetitiveWindowDefinitionError(name: String, ctx: WindowClauseContext): Throwable = {
@@ -135,7 +153,7 @@ object QueryParsingErrors extends QueryErrorsBase {
   }
 
   def naturalCrossJoinUnsupportedError(ctx: RelationContext): Throwable = {
-    new ParseException("UNSUPPORTED_FEATURE", Array("NATURAL CROSS JOIN."), ctx)
+    new ParseException("UNSUPPORTED_FEATURE", Array(toSQLStmt("NATURAL CROSS JOIN") + "."), ctx)
   }
 
   def emptyInputForTableSampleError(ctx: ParserRuleContext): Throwable = {
@@ -301,15 +319,18 @@ object QueryParsingErrors extends QueryErrorsBase {
   }
 
   def showFunctionsUnsupportedError(identifier: String, ctx: IdentifierContext): Throwable = {
-    new ParseException("INVALID_SQL_SYNTAX",
-      Array(s"SHOW ${toSQLId(identifier)} FUNCTIONS not supported"), ctx)
+    new ParseException(
+      errorClass = "INVALID_SQL_SYNTAX",
+      messageParameters = Array(
+        s"${toSQLStmt("SHOW")} ${toSQLId(identifier)} ${toSQLStmt("FUNCTIONS")} not supported"),
+      ctx)
   }
 
   def showFunctionsInvalidPatternError(pattern: String, ctx: ParserRuleContext): Throwable = {
     new ParseException(
       errorClass = "INVALID_SQL_SYNTAX",
       messageParameters = Array(
-        s"Invalid pattern in SHOW FUNCTIONS: ${toSQLId(pattern)}. " +
+        s"Invalid pattern in ${toSQLStmt("SHOW FUNCTIONS")}: ${toSQLId(pattern)}. " +
         s"It must be a ${toSQLType(StringType)} literal."),
       ctx)
   }
@@ -416,13 +437,20 @@ object QueryParsingErrors extends QueryErrorsBase {
   }
 
   def createFuncWithBothIfNotExistsAndReplaceError(ctx: CreateFunctionContext): Throwable = {
-    new ParseException("INVALID_SQL_SYNTAX",
-      Array("CREATE FUNCTION with both IF NOT EXISTS and REPLACE is not allowed."), ctx)
+    new ParseException(
+      errorClass = "INVALID_SQL_SYNTAX",
+      messageParameters = Array(
+        s"${toSQLStmt("CREATE FUNCTION")} with both ${toSQLStmt("IF NOT EXISTS")} " +
+        s"and ${toSQLStmt("REPLACE")} is not allowed."),
+      ctx)
   }
 
   def defineTempFuncWithIfNotExistsError(ctx: CreateFunctionContext): Throwable = {
-    new ParseException("INVALID_SQL_SYNTAX",
-      Array("It is not allowed to define a TEMPORARY function with IF NOT EXISTS."), ctx)
+    new ParseException(
+      errorClass = "INVALID_SQL_SYNTAX",
+      messageParameters = Array(
+        s"It is not allowed to define a ${toSQLStmt("TEMPORARY FUNCTION")}" +
+        s" with ${toSQLStmt("IF NOT EXISTS")}."), ctx)
   }
 
   def unsupportedFunctionNameError(funcName: Seq[String], ctx: CreateFunctionContext): Throwable = {
@@ -435,7 +463,8 @@ object QueryParsingErrors extends QueryErrorsBase {
       ctx: CreateFunctionContext): Throwable = {
     new ParseException(
       "INVALID_SQL_SYNTAX",
-      Array("Specifying a database in CREATE TEMPORARY FUNCTION is not allowed: " +
+      Array(
+        s"Specifying a database in ${toSQLStmt("CREATE TEMPORARY FUNCTION")} is not allowed: " +
         toSQLId(databaseName)),
       ctx)
   }
@@ -449,8 +478,12 @@ object QueryParsingErrors extends QueryErrorsBase {
   }
 
   def invalidNameForDropTempFunc(name: Seq[String], ctx: ParserRuleContext): Throwable = {
-    new ParseException("INVALID_SQL_SYNTAX",
-      Array(s"DROP TEMPORARY FUNCTION requires a single part name but got: ${toSQLId(name)}"), ctx)
+    new ParseException(
+      errorClass = "INVALID_SQL_SYNTAX",
+      messageParameters = Array(
+        s"${toSQLStmt("DROP TEMPORARY FUNCTION")} requires a single part name but got: " +
+        toSQLId(name)),
+      ctx)
   }
 
   def defaultColumnNotImplementedYetError(ctx: ParserRuleContext): Throwable = {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ExtractPythonUDFFromJoinConditionSuite.scala

@@ -187,9 +187,9 @@ class ExtractPythonUDFFromJoinConditionSuite extends PlanTest {
           condition = Some(unevaluableJoinCond))
         Optimize.execute(query.analyze)
       }
-      assert(e.message.contentEquals(
+      assert(e.message ==
         "The feature is not supported: " +
-        s"Using PythonUDF in join condition of join type $joinType is not supported"))
+        s"""Using PythonUDF in join condition of join type "${joinType.sql}" is not supported.""")
 
       val query2 = testRelationLeft.join(
         testRelationRight,

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala

@@ -1250,7 +1250,7 @@ class PlanParserSuite extends AnalysisTest {
         |    "escapeChar" = "\\")
         |FROM testData
       """.stripMargin,
-      "TRANSFORM with serde is only supported in hive mode")
+      "\"TRANSFORM\" with serde is only supported in hive mode")
   }
 
 

sql/core/src/test/resources/sql-tests/results/describe.sql.out

@@ -462,7 +462,7 @@ DESC temp_v PARTITION (c='Us', d=1)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-The operation 'DESC PARTITION' is not allowed on the temporary view: `temp_v`
+The operation "DESC PARTITION" is not allowed on the temporary view: `temp_v`
 
 
 -- !query
@@ -539,7 +539,7 @@ DESC v PARTITION (c='Us', d=1)
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-The operation 'DESC PARTITION' is not allowed on the view: `v`
+The operation "DESC PARTITION" is not allowed on the view: `v`
 
 
 -- !query

sql/core/src/test/resources/sql-tests/results/join-lateral.sql.out

@@ -153,7 +153,7 @@ struct<>
 -- !query output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-The feature is not supported: LATERAL join with NATURAL join.(line 1, pos 14)
+The feature is not supported: "LATERAL" join with "NATURAL" join.(line 1, pos 14)
 
 == SQL ==
 SELECT * FROM t1 NATURAL JOIN LATERAL (SELECT c1 + c2 AS c2)
@@ -167,7 +167,7 @@ struct<>
 -- !query output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-The feature is not supported: LATERAL join with USING join.(line 1, pos 14)
+The feature is not supported: "LATERAL" join with "USING" join.(line 1, pos 14)
 
 == SQL ==
 SELECT * FROM t1 JOIN LATERAL (SELECT c1 + c2 AS c2) USING (c2)

sql/core/src/test/resources/sql-tests/results/transform.sql.out

@@ -719,7 +719,7 @@ struct<>
 -- !query output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-The feature is not supported: TRANSFORM does not support DISTINCT/ALL in inputs(line 1, pos 17)
+The feature is not supported: "TRANSFORM" does not support "DISTINCT"/"ALL" in inputs(line 1, pos 17)
 
 == SQL ==
 SELECT TRANSFORM(DISTINCT b, a, c)
@@ -739,7 +739,7 @@ struct<>
 -- !query output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-The feature is not supported: TRANSFORM does not support DISTINCT/ALL in inputs(line 1, pos 17)
+The feature is not supported: "TRANSFORM" does not support "DISTINCT"/"ALL" in inputs(line 1, pos 17)
 
 == SQL ==
 SELECT TRANSFORM(ALL b, a, c)

sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsDSv2Suite.scala

@@ -52,7 +52,7 @@ class QueryCompilationErrorsDSv2Suite
 
         checkAnswer(spark.table(tbl), spark.emptyDataFrame)
         assert(e.getMessage === "The feature is not supported: " +
-          s"IF NOT EXISTS for the table `testcat`.`ns1`.`ns2`.`tbl` by INSERT INTO.")
+          s""""IF NOT EXISTS" for the table `testcat`.`ns1`.`ns2`.`tbl` by "INSERT INTO".""")
         assert(e.getErrorClass === "UNSUPPORTED_FEATURE")
         assert(e.getSqlState === "0A000")
       }

sql/core/src/test/scala/org/apache/spark/sql/errors/QueryCompilationErrorsSuite.scala

@@ -155,7 +155,7 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
     assert(e.getSqlState === "0A000")
     assert(e.message ===
       "The feature is not supported: " +
-      "Using PythonUDF in join condition of join type LeftOuter is not supported")
+      "Using PythonUDF in join condition of join type \"LEFT OUTER\" is not supported.")
   }
 
   test("UNSUPPORTED_FEATURE: Using pandas UDF aggregate expression with pivot") {
@@ -333,7 +333,8 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
         )
         assert(e.getErrorClass === "FORBIDDEN_OPERATION")
         assert(e.message ===
-          s"The operation 'DESC PARTITION' is not allowed on the temporary view: `$tempViewName`")
+          s"""The operation "DESC PARTITION" is not allowed """ +
+          s"on the temporary view: `$tempViewName`")
       }
     }
   }
@@ -358,7 +359,8 @@ class QueryCompilationErrorsSuite extends QueryTest with SharedSparkSession {
         )
         assert(e.getErrorClass === "FORBIDDEN_OPERATION")
         assert(e.message ===
-          s"The operation 'DESC PARTITION' is not allowed on the view: `$viewName`")
+          s"""The operation "DESC PARTITION" is not allowed """ +
+          s"on the view: `$viewName`")
       }
     }
   }

sql/core/src/test/scala/org/apache/spark/sql/errors/QueryExecutionErrorsSuite.scala

@@ -163,7 +163,7 @@ class QueryExecutionErrorsSuite extends QueryTest
     }
     assert(e1.getErrorClass === "UNSUPPORTED_FEATURE")
     assert(e1.getSqlState === "0A000")
-    assert(e1.getMessage === "The feature is not supported: Repeated pivots.")
+    assert(e1.getMessage === """The feature is not supported: Repeated "PIVOT"s.""")
 
     val e2 = intercept[SparkUnsupportedOperationException] {
       trainingSales
@@ -174,7 +174,7 @@ class QueryExecutionErrorsSuite extends QueryTest
     }
     assert(e2.getErrorClass === "UNSUPPORTED_FEATURE")
     assert(e2.getSqlState === "0A000")
-    assert(e2.getMessage === "The feature is not supported: Pivot not after a groupBy.")
+    assert(e2.getMessage === """The feature is not supported: "PIVOT" not after a "GROUP BY".""")
   }
 
   test("INCONSISTENT_BEHAVIOR_CROSS_VERSION: " +