[SPARK-40811][SQL][TESTS] Use checkError() to intercept ParseException: phase 2

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisTest.scala

@@ -223,8 +223,7 @@ trait AnalysisTest extends PlanTest {
     }
   }
 
-  protected def parseException(parser: String => Any)(
-    sqlText: String): ParseException = {
+  protected def parseException(parser: String => Any)(sqlText: String): ParseException = {
     intercept[ParseException](parser(sqlText))
   }
 }

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

@@ -142,22 +142,25 @@ class ParserUtilsSuite extends SparkFunSuite {
 
   test("operationNotAllowed") {
     val errorMessage = "parse.fail.operation.not.allowed.error.message"
-    val e = intercept[ParseException] {
-      operationNotAllowed(errorMessage, showFuncContext)
-    }.getMessage
-    assert(e.contains("Operation not allowed"))
-    assert(e.contains(errorMessage))
+    checkError(
+      exception = intercept[ParseException] {
+        operationNotAllowed(errorMessage, showFuncContext)
+      },
+      errorClass = "_LEGACY_ERROR_TEMP_0035",
+      parameters = Map("message" -> errorMessage))
   }
 
   test("checkDuplicateKeys") {
     val properties = Seq(("a", "a"), ("b", "b"), ("c", "c"))
     checkDuplicateKeys[String](properties, createDbContext)
 
     val properties2 = Seq(("a", "a"), ("b", "b"), ("a", "c"))
-    val e = intercept[ParseException] {
-      checkDuplicateKeys(properties2, createDbContext)
-    }.getMessage
-    assert(e.contains("Found duplicate keys"))
+    checkError(
+      exception = intercept[ParseException] {
+        checkDuplicateKeys(properties2, createDbContext)
+      },
+      errorClass = "DUPLICATE_KEY",
+      parameters = Map("keyColumn" -> "`a`"))
   }
 
   test("source") {
@@ -201,10 +204,12 @@ class ParserUtilsSuite extends SparkFunSuite {
     val message = "ParserRuleContext should not be empty."
     validate(f1(showFuncContext), message, showFuncContext)
 
-    val e = intercept[ParseException] {
-      validate(f1(emptyContext), message, emptyContext)
-    }.getMessage
-    assert(e.contains(message))
+    checkError(
+      exception = intercept[ParseException] {
+        validate(f1(emptyContext), message, emptyContext)
+      },
+      errorClass = "_LEGACY_ERROR_TEMP_0064",
+      parameters = Map("msg" -> message))
   }
 
   test("withOrigin") {

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.parser
 
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkFunSuite, SparkThrowable}
 import org.apache.spark.sql.types._
 
 class TableSchemaParserSuite extends SparkFunSuite {
@@ -30,9 +30,6 @@ class TableSchemaParserSuite extends SparkFunSuite {
     }
   }
 
-  def assertError(sql: String): Unit =
-    intercept[ParseException](CatalystSqlParser.parseTableSchema(sql))
-
   checkTableSchema("a int", new StructType().add("a", "int"))
   checkTableSchema("A int", new StructType().add("A", "int"))
   checkTableSchema("a INT", new StructType().add("a", "int"))
@@ -73,11 +70,31 @@ class TableSchemaParserSuite extends SparkFunSuite {
 
   // Negative cases
   test("Negative cases") {
-    assertError("")
-    assertError("a")
-    assertError("a INT b long")
-    assertError("a INT,, b long")
-    assertError("a INT, b long,,")
-    assertError("a INT, b long, c int,")
+    def parseException(sql: String): SparkThrowable =
+      intercept[ParseException](CatalystSqlParser.parseTableSchema(sql))
+
+    checkError(
+      exception = parseException(""),
+      errorClass = "PARSE_EMPTY_STATEMENT")
+    checkError(
+      exception = parseException("a"),
+      errorClass = "PARSE_SYNTAX_ERROR",
+      parameters = Map("error" -> "end of input", "hint" -> ""))
+    checkError(
+      exception = parseException("a INT b long"),
+      errorClass = "PARSE_SYNTAX_ERROR",
+      parameters = Map("error" -> "'b'", "hint" -> ""))
+    checkError(
+      exception = parseException("a INT,, b long"),
+      errorClass = "PARSE_SYNTAX_ERROR",
+      parameters = Map("error" -> "','", "hint" -> ": extra input ','"))
+    checkError(
+      exception = parseException("a INT, b long,,"),
+      errorClass = "PARSE_SYNTAX_ERROR",
+      parameters = Map("error" -> "','", "hint" -> ""))
+    checkError(
+      exception = parseException("a INT, b long, c int,"),
+      errorClass = "PARSE_SYNTAX_ERROR",
+      parameters = Map("error" -> "end of input", "hint" -> ""))
   }
 }

sql/core/src/test/scala/org/apache/spark/sql/MiscFunctionsSuite.scala

@@ -54,10 +54,10 @@ class MiscFunctionsSuite extends QueryTest with SharedSparkSession {
       SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> "true") {
       Seq("user", "current_user").foreach { func =>
         checkAnswer(sql(s"select $func"), Row(user))
-      }
-      Seq("user()", "current_user()").foreach { func =>
-        val e = intercept[ParseException](sql(s"select $func"))
-        assert(e.getMessage.contains(func))
+        checkError(
+          exception = intercept[ParseException](sql(s"select $func()")),
+          errorClass = "PARSE_SYNTAX_ERROR",
+          parameters = Map("error" -> s"'$func'", "hint" -> ""))
       }
     }
   }

sql/hive/src/test/scala/org/apache/spark/sql/hive/InsertSuite.scala

@@ -706,40 +706,35 @@ class InsertSuite extends QueryTest with TestHiveSingleton with BeforeAndAfter
   test("insert overwrite to dir with mixed syntax") {
     withTempView("test_insert_table") {
       spark.range(10).selectExpr("id", "id AS str").createOrReplaceTempView("test_insert_table")
-
-      val e = intercept[ParseException] {
-        sql(
-          s"""
-             |INSERT OVERWRITE DIRECTORY 'file://tmp'
+      checkError(
+        exception = intercept[ParseException] { sql(
+          s"""INSERT OVERWRITE DIRECTORY 'file://tmp'
              |USING json
              |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
-             |SELECT * FROM test_insert_table
-           """.stripMargin)
-      }.getMessage
-
-      assert(e.contains("Syntax error at or near 'ROW'"))
+             |SELECT * FROM test_insert_table""".stripMargin)
+        },
+        errorClass = "PARSE_SYNTAX_ERROR",
+        parameters = Map("error" -> "'ROW'", "hint" -> ""))
     }
   }
 
   test("insert overwrite to dir with multi inserts") {
     withTempView("test_insert_table") {
       spark.range(10).selectExpr("id", "id AS str").createOrReplaceTempView("test_insert_table")
-
-      val e = intercept[ParseException] {
-        sql(
-          s"""
-             |INSERT OVERWRITE DIRECTORY 'file://tmp2'
-             |USING json
-             |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
-             |SELECT * FROM test_insert_table
-             |INSERT OVERWRITE DIRECTORY 'file://tmp2'
-             |USING json
-             |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
-             |SELECT * FROM test_insert_table
-           """.stripMargin)
-      }.getMessage
-
-      assert(e.contains("Syntax error at or near 'ROW'"))
+      checkError(
+        exception = intercept[ParseException] {
+          sql(
+            s"""INSERT OVERWRITE DIRECTORY 'file://tmp2'
+               |USING json
+               |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
+               |SELECT * FROM test_insert_table
+               |INSERT OVERWRITE DIRECTORY 'file://tmp2'
+               |USING json
+               |ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
+               |SELECT * FROM test_insert_table""".stripMargin)
+        },
+        errorClass = "PARSE_SYNTAX_ERROR",
+        parameters = Map("error" -> "'ROW'", "hint" -> ""))
     }
   }
 

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala

@@ -2678,27 +2678,30 @@ class HiveDDLSuite
   }
 
   test("Hive CTAS can't create partitioned table by specifying schema") {
-    val err1 = intercept[ParseException] {
-      spark.sql(
-        s"""
-           |CREATE TABLE t (a int)
-           |PARTITIONED BY (b string)
-           |STORED AS parquet
-           |AS SELECT 1 as a, "a" as b
-                 """.stripMargin)
-    }.getMessage
-    assert(err1.contains("Schema may not be specified in a Create Table As Select"))
-
-    val err2 = intercept[ParseException] {
-      spark.sql(
-        s"""
-           |CREATE TABLE t
-           |PARTITIONED BY (b string)
-           |STORED AS parquet
-           |AS SELECT 1 as a, "a" as b
-                 """.stripMargin)
-    }.getMessage
-    assert(err2.contains("Partition column types may not be specified in Create Table As Select"))
+    val sql1 =
+      s"""CREATE TABLE t (a int)
+         |PARTITIONED BY (b string)
+         |STORED AS parquet
+         |AS SELECT 1 as a, "a" as b""".stripMargin
+    checkError(
+      exception = intercept[ParseException](sql(sql1)),
+      errorClass = "_LEGACY_ERROR_TEMP_0035",
+      parameters = Map(
+        "message" -> "Schema may not be specified in a Create Table As Select (CTAS) statement"),
+      context = ExpectedContext(sql1, 0, 92))
+
+    val sql2 =
+      s"""CREATE TABLE t
+         |PARTITIONED BY (b string)
+         |STORED AS parquet
+         |AS SELECT 1 as a, "a" as b""".stripMargin
+    checkError(
+      exception = intercept[ParseException](sql(sql2)),
+      errorClass = "_LEGACY_ERROR_TEMP_0035",
+      parameters = Map(
+        "message" ->
+          "Partition column types may not be specified in Create Table As Select (CTAS)"),
+      context = ExpectedContext(sql2, 0, 84))
   }
 
   test("Hive CTAS with dynamic partition") {

sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala

@@ -1744,17 +1744,21 @@ abstract class SQLQuerySuiteBase extends QueryTest with SQLTestUtils with TestHi
 
   test("SPARK-14981: DESC not supported for sorting columns") {
     withTable("t") {
-      val cause = intercept[ParseException] {
-        sql(
-          """CREATE TABLE t USING PARQUET
-            |OPTIONS (PATH '/path/to/file')
-            |CLUSTERED BY (a) SORTED BY (b DESC) INTO 2 BUCKETS
-            |AS SELECT 1 AS a, 2 AS b
-          """.stripMargin
-        )
-      }
-
-      assert(cause.getMessage.contains("Column ordering must be ASC, was 'DESC'"))
+      checkError(
+        exception = intercept[ParseException] {
+          sql(
+            """CREATE TABLE t USING PARQUET
+              |OPTIONS (PATH '/path/to/file')
+              |CLUSTERED BY (a) SORTED BY (b DESC) INTO 2 BUCKETS
+              |AS SELECT 1 AS a, 2 AS b
+            """.stripMargin)
+        },
+        errorClass = "_LEGACY_ERROR_TEMP_0035",
+        parameters = Map("message" -> "Column ordering must be ASC, was 'DESC'"),
+        context = ExpectedContext(
+          fragment = "CLUSTERED BY (a) SORTED BY (b DESC) INTO 2 BUCKETS",
+          start = 60,
+          stop = 109))
     }
   }