[SPARK-47158][SQL] Assign proper name and sqlState to _LEGACY_ERROR_TEMP_(2134|2231)

### What changes were proposed in this pull request?

This PR proposes to assign proper name and `sqlState` to `_LEGACY_ERROR_TEMP_(2134|2231)`

- `_LEGACY_ERROR_TEMP_2134` -> `CANNOT_PARSE_STRING_AS_DATATYPE`
- `_LEGACY_ERROR_TEMP_2231` -> `UNSUPPORTED_CALL.FIELD_INDEX`

### Why are the changes needed?

To improve error usability

### Does this PR introduce _any_ user-facing change?

No API changes, but the user-facing error message will be improved

### How was this patch tested?

Added UTs.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#45244 from itholic/TOP_LEGACY_ERRORS.

Authored-by: Haejoon Lee <[email protected]>
Signed-off-by: Max Gekk <[email protected]>

Author: itholic

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

@@ -3735,6 +3735,18 @@
     "message" : [
       "Cannot call the method \"<methodName>\" of the class \"<className>\"."
     ],
+    "subClass" : {
+      "FIELD_INDEX" : {
+        "message" : [
+          "The row shall have a schema to get an index of the field <fieldName>."
+        ]
+      },
+      "WITHOUT_SUGGESTION" : {
+        "message" : [
+          ""
+        ]
+      }
+    },
     "sqlState" : "0A000"
   },
   "UNSUPPORTED_CHAR_OR_VARCHAR_AS_STRING" : {
@@ -6172,11 +6184,6 @@
       "Exception when registering StreamingQueryListener."
     ]
   },
-  "_LEGACY_ERROR_TEMP_2134" : {
-    "message" : [
-      "Cannot parse field value <value> for pattern <pattern> as target spark data type [<dataType>]."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_2138" : {
     "message" : [
       "Cannot have circular references in bean class, but got the circular reference of class <clazz>."
@@ -6517,11 +6524,6 @@
       "Primitive types are not supported."
     ]
   },
-  "_LEGACY_ERROR_TEMP_2231" : {
-    "message" : [
-      "fieldIndex on a Row without schema is undefined."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_2232" : {
     "message" : [
       "Value at index <index> is null."

common/utils/src/main/scala/org/apache/spark/SparkException.scala

@@ -262,7 +262,8 @@ private[spark] object SparkUnsupportedOperationException {
     } else {
       Map("className" -> "?", "methodName" -> "?")
     }
-    new SparkUnsupportedOperationException("UNSUPPORTED_CALL", messageParameters)
+    new SparkUnsupportedOperationException(
+      "UNSUPPORTED_CALL.WITHOUT_SUGGESTION", messageParameters)
   }
 }
 

docs/sql-error-conditions-unsupported-call-error-class.md

@@ -0,0 +1,41 @@
+---
+layout: global
+title: UNSUPPORTED_CALL error class
+displayTitle: UNSUPPORTED_CALL error class
+license: |
+  Licensed to the Apache Software Foundation (ASF) under one or more
+  contributor license agreements.  See the NOTICE file distributed with
+  this work for additional information regarding copyright ownership.
+  The ASF licenses this file to You under the Apache License, Version 2.0
+  (the "License"); you may not use this file except in compliance with
+  the License.  You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+---
+
+<!--
+  DO NOT EDIT THIS FILE.
+  It was generated automatically by `org.apache.spark.SparkThrowableSuite`.
+-->
+
+[SQLSTATE: 0A000](sql-error-conditions-sqlstates.html#class-0A-feature-not-supported)
+
+Cannot call the method "`<methodName>`" of the class "`<className>`".
+
+This error class has the following derived error classes:
+
+## FIELD_INDEX
+
+The row shall have a schema to get an index of the field `<fieldName>`.
+
+## WITHOUT_SUGGESTION
+
+
+
+

docs/sql-error-conditions.md

@@ -2435,12 +2435,14 @@ For more details see [UNSUPPORTED_ADD_FILE](sql-error-conditions-unsupported-add
 
 Unsupported arrow type `<typeName>`.
 
-### UNSUPPORTED_CALL
+### [UNSUPPORTED_CALL](sql-error-conditions-unsupported-call-error-class.html)
 
 [SQLSTATE: 0A000](sql-error-conditions-sqlstates.html#class-0A-feature-not-supported)
 
 Cannot call the method "`<methodName>`" of the class "`<className>`".
 
+For more details see [UNSUPPORTED_CALL](sql-error-conditions-unsupported-call-error-class.html)
+
 ### UNSUPPORTED_CHAR_OR_VARCHAR_AS_STRING
 
 [SQLSTATE: 0A000](sql-error-conditions-sqlstates.html#class-0A-feature-not-supported)

sql/api/src/main/scala/org/apache/spark/sql/Row.scala

@@ -380,7 +380,7 @@ trait Row extends Serializable {
    * @throws IllegalArgumentException when a field `name` does not exist.
    */
   def fieldIndex(name: String): Int = {
-    throw DataTypeErrors.fieldIndexOnRowWithoutSchemaError()
+    throw DataTypeErrors.fieldIndexOnRowWithoutSchemaError(fieldName = name)
   }
 
   /**

sql/api/src/main/scala/org/apache/spark/sql/errors/DataTypeErrors.scala

@@ -261,8 +261,14 @@ private[sql] object DataTypeErrors extends DataTypeErrorsBase {
       messageParameters = Map("raw" -> s"'$raw'"))
   }
 
-  def fieldIndexOnRowWithoutSchemaError(): SparkUnsupportedOperationException = {
-    new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_2231")
+  def fieldIndexOnRowWithoutSchemaError(fieldName: String): SparkUnsupportedOperationException = {
+    new SparkUnsupportedOperationException(
+      errorClass = "UNSUPPORTED_CALL.FIELD_INDEX",
+      messageParameters = Map(
+        "methodName" -> "fieldIndex",
+        "className" -> "Row",
+        "fieldName" -> toSQLId(fieldName))
+    )
   }
 
   def valueIsNullError(index: Int): Throwable = {

sql/api/src/main/scala/org/apache/spark/sql/errors/ExecutionErrors.scala

@@ -127,13 +127,10 @@ private[sql] trait ExecutionErrors extends DataTypeErrorsBase {
   }
 
   def cannotParseStringAsDataTypeError(pattern: String, value: String, dataType: DataType)
-  : SparkRuntimeException = {
-    new SparkRuntimeException(
-      errorClass = "_LEGACY_ERROR_TEMP_2134",
-      messageParameters = Map(
-        "value" -> toSQLValue(value),
-        "pattern" -> toSQLValue(pattern),
-        "dataType" -> dataType.toString))
+  : Throwable = {
+    SparkException.internalError(
+      s"Cannot parse field value ${toSQLValue(value)} for pattern ${toSQLValue(pattern)} " +
+        s"as the target spark data type ${toSQLType(dataType)}.")
   }
 
   def unsupportedArrowTypeError(typeName: ArrowType): SparkUnsupportedOperationException = {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala

@@ -474,7 +474,7 @@ class ShuffleSpecSuite extends SparkFunSuite with SQLHelper {
       exception = intercept[SparkUnsupportedOperationException] {
         RangeShuffleSpec(10, distribution).createPartitioning(distribution.clustering)
       },
-      errorClass = "UNSUPPORTED_CALL",
+      errorClass = "UNSUPPORTED_CALL.WITHOUT_SUGGESTION",
       parameters = Map(
         "methodName" -> "createPartitioning$",
         "className" -> "org.apache.spark.sql.catalyst.plans.physical.ShuffleSpec"))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/TimestampFormatterSuite.scala

@@ -17,11 +17,13 @@
 
 package org.apache.spark.sql.catalyst.util
 
-import java.time.{DateTimeException, LocalDateTime}
+import java.time.{DateTimeException, LocalDateTime, ZoneId}
+import java.util.Locale
 
-import org.apache.spark.SparkUpgradeException
+import org.apache.spark.{SparkException, SparkUpgradeException}
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
 import org.apache.spark.sql.catalyst.util.DateTimeUtils._
+import org.apache.spark.sql.catalyst.util.LegacyDateFormats.LENIENT_SIMPLE_DATE_FORMAT
 import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -502,4 +504,23 @@ class TimestampFormatterSuite extends DatetimeFormatterSuite {
     assert(formatter.parseOptional("9999-12-31 23:59:59.999").isEmpty)
     assert(formatter.parseWithoutTimeZoneOptional("9999-12-31 23:59:59.999", true).isEmpty)
   }
+
+  test("fail to parse string as TimestampNTZ with invalid format") {
+    val zoneId = ZoneId.systemDefault()
+    val locale = Locale.getDefault()
+    val formatter = new DefaultTimestampFormatter(
+      zoneId, locale, LENIENT_SIMPLE_DATE_FORMAT, isParsing = true)
+
+    val invalidTimestampStr = "2021-13-01T25:61:61"
+
+    checkError(
+      exception = intercept[SparkException] {
+        formatter.parseWithoutTimeZone(invalidTimestampStr, allowTimeZone = false)
+      },
+      errorClass = "INTERNAL_ERROR",
+      parameters = Map(
+        "message" -> ("Cannot parse field value '2021-13-01T25:61:61' for pattern " +
+          "'yyyy-MM-dd HH:mm:ss' as the target spark data type \"TIMESTAMP_NTZ\"."))
+    )
+  }
 }

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql
 
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkFunSuite, SparkUnsupportedOperationException}
 import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, SpecificInternalRow}
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types._
@@ -111,4 +111,16 @@ class RowSuite extends SparkFunSuite with SharedSparkSession {
     assert(Row(Seq()).getSeq(0) === List())
     assert(Row(null).getSeq(0) === null)
   }
+
+  test("access fieldIndex on Row without schema") {
+    val rowWithoutSchema = Row(1, "foo", 3.14)
+
+    checkError(
+      exception = intercept[SparkUnsupportedOperationException] {
+        rowWithoutSchema.fieldIndex("foo")
+      },
+      errorClass = "UNSUPPORTED_CALL.FIELD_INDEX",
+      parameters = Map("methodName" -> "fieldIndex", "className" -> "Row", "fieldName" -> "`foo`")
+    )
+  }
 }

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

@@ -958,7 +958,7 @@ class QueryCompilationErrorsSuite
       exception = intercept[SparkUnsupportedOperationException] {
         new UnsafeRow(1).update(0, 1)
       },
-      errorClass = "UNSUPPORTED_CALL",
+      errorClass = "UNSUPPORTED_CALL.WITHOUT_SUGGESTION",
       parameters = Map(
         "methodName" -> "update",
         "className" -> "org.apache.spark.sql.catalyst.expressions.UnsafeRow"))