[SPARK-41179][SQL] Assign a name to the error class _LEGACY_ERROR_TEMP_1092

core/src/main/resources/error/error-classes.json

@@ -751,6 +751,11 @@
       "<protobufClassName> is not a Protobuf message type"
     ]
   },
+  "INVALID_SCHEMA" : {
+    "message" : [
+      "The expression <expr> is not a valid schema string."
+    ]
+  },
   "INVALID_SQL_SYNTAX" : {
     "message" : [
       "Invalid SQL syntax: <inputString>"
@@ -2165,11 +2170,6 @@
       "Cannot read table property '<key>' as it's corrupted.<details>."
     ]
   },
-  "_LEGACY_ERROR_TEMP_1092" : {
-    "message" : [
-      "The expression '<expr>' is not a valid schema string."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_1093" : {
     "message" : [
       "Schema should be specified in DDL format as a string literal or output of the schema_of_json/schema_of_csv functions instead of <expr>."

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

@@ -996,8 +996,8 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
 
   def invalidSchemaStringError(exp: Expression): Throwable = {
     new AnalysisException(
-      errorClass = "_LEGACY_ERROR_TEMP_1092",
-      messageParameters = Map("expr" -> exp.sql))
+      errorClass = "INVALID_SCHEMA",
+      messageParameters = Map("expr" -> toSQLExpr(exp)))
   }
 
   def schemaNotFoldableError(exp: Expression): Throwable = {

sql/core/src/test/resources/sql-tests/results/csv-functions.sql.out

@@ -22,9 +22,9 @@ struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "_LEGACY_ERROR_TEMP_1092",
+  "errorClass" : "INVALID_SCHEMA",
   "messageParameters" : {
-    "expr" : "1"
+    "expr" : "\"1\""
   },
   "queryContext" : [ {
     "objectType" : "",

sql/core/src/test/resources/sql-tests/results/json-functions.sql.out

@@ -148,9 +148,9 @@ struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
 {
-  "errorClass" : "_LEGACY_ERROR_TEMP_1092",
+  "errorClass" : "INVALID_SCHEMA",
   "messageParameters" : {
-    "expr" : "1"
+    "expr" : "\"1\""
   },
   "queryContext" : [ {
     "objectType" : "",

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

@@ -363,10 +363,13 @@ class CsvFunctionsSuite extends QueryTest with SharedSparkSession {
     }.getMessage
     assert(errMsg.contains("Schema should be specified in DDL format as a string literal"))
 
-    val errMsg2 = intercept[AnalysisException] {
-      Seq("1").toDF("csv").select(from_csv($"csv", lit(1), options)).collect()
-    }.getMessage
-    assert(errMsg2.contains("The expression '1' is not a valid schema string"))
+    checkError(
+      exception = intercept[AnalysisException] {
+        Seq("1").toDF("csv").select(from_csv($"csv", lit(1), options)).collect()
+      },
+      errorClass = "INVALID_SCHEMA",
+      parameters = Map("expr" -> "\"1\"")
+    )
   }
 
   test("schema_of_csv - infers the schema of foldable CSV string") {

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

@@ -5179,6 +5179,23 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSparkSession {
       Seq(Row(Map("a" -> Map("a" -> 6, "b" -> 8), "b" -> Map("a" -> 8, "b" -> 10))))
     )
   }
+
+  test("from_json - invalid schema string") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        sql("select from_json('{\"a\":1}', 1)")
+      },
+      errorClass = "INVALID_SCHEMA",
+      parameters = Map(
+        "expr" -> "\"1\""
+      ),
+      context = ExpectedContext(
+        fragment = "from_json('{\"a\":1}', 1)",
+        start = 7,
+        stop = 29
+      )
+    )
+  }
 }
 
 object DataFrameFunctionsSuite {

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

@@ -424,10 +424,19 @@ class JsonFunctionsSuite extends QueryTest with SharedSparkSession {
         "from_json(value, 'time Timestamp', map('timestampFormat', 'dd/MM/yyyy HH:mm'))"),
       Row(Row(java.sql.Timestamp.valueOf("2015-08-26 18:00:00.0"))))
 
-    val errMsg1 = intercept[AnalysisException] {
-      df3.selectExpr("from_json(value, 1)")
-    }
-    assert(errMsg1.getMessage.startsWith("The expression '1' is not a valid schema string"))
+    checkError(
+      exception = intercept[AnalysisException] {
+        df3.selectExpr("from_json(value, 1)")
+      },
+      errorClass = "INVALID_SCHEMA",
+      parameters = Map("expr" -> "\"1\""),
+      context = ExpectedContext(
+        fragment = "from_json(value, 1)",
+        start = 0,
+        stop = 18
+      )
+    )
+
     val errMsg2 = intercept[AnalysisException] {
       df3.selectExpr("""from_json(value, 'time InvalidType')""")
     }