[SPARK-58945][SQL] Fix mismatched messageParameters keys that cause INTERNAL_ERROR

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

@@ -1838,7 +1838,7 @@
   },
   "CURSOR_OUTSIDE_SCRIPT" : {
     "message" : [
-      "Cursor operations can only be used within SQL scripts."
+      "Cursor <cursorName> can only be used within SQL scripts."
     ],
     "sqlState" : "0A000"
   },
@@ -11818,7 +11818,7 @@
   },
   "_LEGACY_ERROR_TEMP_3070" : {
     "message" : [
-      "<internalName> is a reserved column name that cannot be read in combination with <colName> column."
+      "Unrecognized file metadata field: <field>"
     ]
   },
   "_LEGACY_ERROR_TEMP_3071" : {

sql/api/src/main/scala/org/apache/spark/sql/catalyst/analysis/noSuchItemsExceptions.scala

@@ -23,7 +23,7 @@ import org.apache.spark.sql.catalyst.util.QuotingUtils.{quoted, quoteIdentifier,
 import org.apache.spark.sql.connector.catalog.Identifier
 import org.apache.spark.util.ArrayImplicits._
 
-private[analysis] object NoSuchItemExceptionHelper {
+private[sql] object NoSuchItemExceptionHelper {
 
   /** Format a search path for TABLE_OR_VIEW_NOT_FOUND (e.g. [`cat`.`ns`]). */
   def formatSearchPath(searchPath: Seq[String]): String =

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCursors.scala

@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.{CursorReference, UnresolvedCur
 import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_CURSOR
+import org.apache.spark.sql.errors.DataTypeErrorsBase
 import org.apache.spark.sql.internal.SQLConf
 
 /**
@@ -35,7 +36,7 @@ import org.apache.spark.sql.internal.SQLConf
  * 3. Looks up the cursor definition from the scripting context
  * 4. Fails early if cursor is not found
  */
-class ResolveCursors extends Rule[LogicalPlan] {
+class ResolveCursors extends Rule[LogicalPlan] with DataTypeErrorsBase {
 
   override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveExpressionsWithPruning(
     _.containsPattern(UNRESOLVED_CURSOR)) {
@@ -82,7 +83,7 @@ class ResolveCursors extends Rule[LogicalPlan] {
         // Cursors are only allowed within SQL scripts
         throw new AnalysisException(
           errorClass = "CURSOR_OUTSIDE_SCRIPT",
-          messageParameters = Map("cursorName" -> nameParts.mkString(".")))
+          messageParameters = Map("cursorName" -> toSQLId(nameParts)))
       }
 
     // Use the SqlScriptingExecutionContextExtension API for cursor lookup

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala

@@ -125,7 +125,7 @@ class CSVOptions(
   val extension = {
     val ext = parameters.getOrElse(EXTENSION, "csv")
     if (ext.size != 3 && !ext.forall(_.isLetter)) {
-      throw QueryExecutionErrors.invalidFileExtensionError(EXTENSION, ext)
+      throw QueryExecutionErrors.invalidFileExtensionError("csv", ext)
     }
 
     ext

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxCountDistinctForIntervals.scala

@@ -86,7 +86,8 @@ case class ApproxCountDistinctForIntervals(
         errorSubClass = "NON_FOLDABLE_INPUT",
         messageParameters = Map(
           "inputName" -> toSQLId("endpointsExpression"),
-          "inputType" -> toSQLType(endpointsExpression.dataType)))
+          "inputType" -> toSQLType(endpointsExpression.dataType),
+          "inputExpr" -> toSQLExpr(endpointsExpression)))
     } else {
       endpointsExpression.dataType match {
         case ArrayType(_: NumericType | DateType | TimestampType | TimestampNTZType |

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

@@ -4593,7 +4593,10 @@ class AstBuilder extends DataTypeAstBuilder
     val path = if (field.startsWith("[")) "$" + field else s"$$.$field"
     val parsedPath = JsonPathParser.parse(path)
     if (parsedPath.isEmpty) {
-      throw new ParseException(errorClass = "PARSE_SYNTAX_ERROR", ctx = ctx)
+      throw new ParseException(
+        errorClass = "PARSE_SYNTAX_ERROR",
+        messageParameters = Map("error" -> s"'$field'", "hint" -> ""),
+        ctx = ctx)
     }
     val potentialAlias = parsedPath.get.collect { case Named(name) => name }.lastOption
     val node = SemiStructuredExtract(expression(ctx.col), path)

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

@@ -4749,7 +4749,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
   def invalidUDFClassError(invalidClass: String): Throwable = {
     new InvalidUDFClassException(
       errorClass = "_LEGACY_ERROR_TEMP_2450",
-      messageParameters = Map("invalidClass" -> invalidClass))
+      messageParameters = Map("clazz" -> invalidClass))
   }
 
   def cannotInstantiateHiveFunctionError(clazz: String, e: Throwable): Throwable = {

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

@@ -3234,8 +3234,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       messageParameters = Map(
         "functionName" -> toSQLId(functionName),
         "parameter" -> toSQLId("extension"),
-        "fileExtension" -> toSQLId(extension),
-        "acceptable" -> "Extension is limited to exactly 3 letters (e.g. csv, tsv, etc...)"))
+        "invalidValue" -> toSQLId(extension)))
   }
 
   def invalidCharsetError(functionName: String, charset: String): RuntimeException = {
@@ -3259,7 +3258,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
   def invalidWriterCommitMessageError(details: String): Throwable = {
     new SparkRuntimeException(
       errorClass = "INVALID_WRITER_COMMIT_MESSAGE",
-      messageParameters = Map("details" -> details))
+      messageParameters = Map("detail" -> details))
   }
 
   def codecNotAvailableError(codecName: String, availableCodecs: String): Throwable = {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproxCountDistinctForIntervalsSuite.scala

@@ -75,7 +75,8 @@ class ApproxCountDistinctForIntervalsSuite extends SparkFunSuite {
         errorSubClass = "NON_FOLDABLE_INPUT",
         messageParameters = Map(
           "inputName" -> "`endpointsExpression`",
-          "inputType" -> "\"ARRAY<DOUBLE>\"")))
+          "inputType" -> "\"ARRAY<DOUBLE>\"",
+          "inputExpr" -> "\"array(b)\"")))
 
     wrongEndpoints = ApproxCountDistinctForIntervals(
       AttributeReference("a", DoubleType)(),

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

@@ -291,6 +291,14 @@ class ExpressionParserSuite extends AnalysisTest {
     assertEqual("a is not distinct from b", $"a" <=> $"b")
   }
 
+  test("invalid semi-structured extract path") {
+    checkError(
+      exception = parseException("c:['']"),
+      condition = "PARSE_SYNTAX_ERROR",
+      parameters = Map("error" -> "'['']'", "hint" -> ""),
+      queryContext = Array(ExpectedContext("c:['']", 0, 5)))
+  }
+
   test("binary arithmetic expressions") {
     // Simple operations
     assertEqual("a * b", $"a" * $"b")

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/StateStoreErrors.scala

@@ -398,9 +398,9 @@ class StateStoreColumnFamilyMismatch(
   extends SparkUnsupportedOperationException(
     errorClass = "STATE_STORE_COLUMN_FAMILY_SCHEMA_INCOMPATIBLE",
     messageParameters = Map(
-      "columnFamilyName" -> columnFamilyName,
-      "oldColumnFamilySchema" -> oldColumnFamilySchema,
-      "newColumnFamilySchema" -> newColumnFamilySchema))
+      "colFamilyName" -> columnFamilyName,
+      "oldSchema" -> oldColumnFamilySchema,
+      "newSchema" -> newColumnFamilySchema))
 
 class StatefulProcessorCannotPerformOperationWithInvalidTimeMode(
     operationType: String,

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

@@ -26,7 +26,13 @@ import scala.jdk.CollectionConverters._
 import scala.util.control.NonFatal
 
 import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
-import org.apache.spark.sql.catalyst.analysis.{IndexAlreadyExistsException, NoSuchIndexException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.analysis.{
+  IndexAlreadyExistsException,
+  NoSuchIndexException,
+  NoSuchItemExceptionHelper,
+  NoSuchNamespaceException,
+  NoSuchTableException,
+  TableAlreadyExistsException}
 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
@@ -225,10 +231,14 @@ private[sql] case class H2Dialect() extends JdbcDialect with NoLegacyJDBCError {
               cause = Some(e))
           // TABLE_OR_VIEW_NOT_FOUND_1
           case 42102 =>
-            val relationName = messageParameters.getOrElse("tableName", "")
+            val relationName = messageParameters
+              .getOrElse("tableName", messageParameters.getOrElse("oldName", ""))
             throw new NoSuchTableException(
               errorClass = "TABLE_OR_VIEW_NOT_FOUND",
-              messageParameters = Map("relationName" -> relationName),
+              messageParameters = Map(
+                "relationName" -> relationName,
+                // Use the shared empty-search-path rendering for this dialect-classification path.
+                "searchPath" -> NoSuchItemExceptionHelper.formatSearchPath(Seq.empty)),
               cause = Some(e))
           // SCHEMA_NOT_FOUND_1
           case 90079 =>

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

@@ -23,7 +23,7 @@ import org.apache.spark.{SPARK_DOC_ROOT, SparkIllegalArgumentException, SparkUns
 import org.apache.spark.sql._
 import org.apache.spark.sql.api.java.{UDF1, UDF2, UDF23Test}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType}
+import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable, CatalogTableType, InvalidUDFClassException}
 import org.apache.spark.sql.catalyst.expressions.{Coalesce, Literal, UnsafeRow}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand
@@ -283,6 +283,14 @@ class QueryCompilationErrorsSuite
       sqlState = "0A000")
   }
 
+  test("SPARK-58945: invalid UDF class error reports clazz") {
+    checkError(
+      exception = QueryCompilationErrors.invalidUDFClassError("example.InvalidFunction")
+        .asInstanceOf[InvalidUDFClassException],
+      condition = "_LEGACY_ERROR_TEMP_2450",
+      parameters = Map("clazz" -> "example.InvalidFunction"))
+  }
+
   test("GROUPING_COLUMN_MISMATCH: not found the grouping column") {
     val groupingColMismatchEx = intercept[AnalysisException] {
       courseSales.cube("course", "year").agg(grouping("earnings")).explain()

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

@@ -229,6 +229,31 @@ class QueryExecutionErrorsSuite
       "CBC", "NoPadding")
   }
 
+  test("SPARK-58945: invalid file extension reports invalidValue") {
+    withTempDir { dir =>
+      val path = new File(dir, "data").getCanonicalPath
+      // Use a value that is unambiguously invalid so this test focuses on the
+      // rendered error message rather than the full extension validation matrix.
+      checkError(
+        exception = intercept[SparkIllegalArgumentException] {
+          spark.range(1).write.option("extension", "12").csv(path)
+        },
+        condition = "INVALID_PARAMETER_VALUE.EXTENSION",
+        parameters = Map(
+          "functionName" -> "`csv`",
+          "parameter" -> "`extension`",
+          "invalidValue" -> "`12`"))
+    }
+  }
+
+  test("SPARK-58945: invalid writer commit message reports detail") {
+    checkError(
+      exception = QueryExecutionErrors.invalidWriterCommitMessageError("zero")
+        .asInstanceOf[SparkRuntimeException],
+      condition = "INVALID_WRITER_COMMIT_MESSAGE",
+      parameters = Map("detail" -> "zero"))
+  }
+
   test("UNSUPPORTED_FEATURE: unsupported types (map and struct) in lit()") {
     def checkUnsupportedTypeInLiteral(v: Any, literal: String, dataType: String): Unit = {
       checkError(

sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CursorCommandUtilsSuite.scala

@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.command.v2
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.SqlScriptingContextManager
+
+class CursorCommandUtilsSuite extends SparkFunSuite {
+
+  test("SPARK-58945: CursorCommandUtils reports cursor name outside scripts") {
+    assert(SqlScriptingContextManager.get().isEmpty)
+    checkError(
+      exception = intercept[AnalysisException] {
+        CursorCommandUtils.getScriptingContext("cur")
+      },
+      condition = "CURSOR_OUTSIDE_SCRIPT",
+      parameters = Map("cursorName" -> "`cur`"))
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceCustomMetadataStructSuite.scala

@@ -21,7 +21,7 @@ import java.io.File
 
 import org.apache.hadoop.fs.{FileStatus, Path}
 
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.{AnalysisException, Row}
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Expression, FileSourceConstantMetadataStructField, FileSourceGeneratedMetadataStructField, Literal}
 import org.apache.spark.sql.catalyst.util.GenericArrayData
@@ -413,6 +413,21 @@ class FileSourceCustomMetadataStructSuite extends SharedSparkSession {
           Row(1, 112L, 1L, f1.getLen, f1.getPath.getName)))
     }
   }
+
+  test("SPARK-58945: invalid file metadata fields report the field") {
+    withTempData("parquet", FILE_SCHEMA) { (_, f0, f1) =>
+      val invalidField = StructField("bad", StringType)
+      val format = new TestFileFormat(Seq(invalidField))
+      val df = createDF(format, Seq(FileStatusWithMetadata(f0), FileStatusWithMetadata(f1)))
+
+      checkError(
+        exception = intercept[AnalysisException] {
+          df.select("_metadata.bad").collect()
+        },
+        condition = "_LEGACY_ERROR_TEMP_3070",
+        parameters = Map("field" -> invalidField.toString))
+    }
+  }
 }
 
 object FileSourceCustomMetadataStructSuite {

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/jdbc/JDBCTableCatalogSuite.scala

@@ -25,7 +25,7 @@ import org.apache.logging.log4j.Level
 
 import org.apache.spark.{SparkConf, SparkIllegalArgumentException, SparkRuntimeException}
 import org.apache.spark.sql.{AnalysisException, Row}
-import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, TableAlreadyExistsException}
+import org.apache.spark.sql.catalyst.analysis.{NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.util.CharVarcharUtils
 import org.apache.spark.sql.connector.catalog.{Identifier, TableSummary}
@@ -229,6 +229,18 @@ class JDBCTableCatalogSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-58945: H2 renameTable reports source table when it is missing") {
+    val e = intercept[NoSuchTableException] {
+      tableCatalog.renameTable(
+        Identifier.of(Array("test"), "not_existing_table"),
+        Identifier.of(Array("test"), "dst_table"))
+    }
+    checkErrorTableNotFoundWithSearchPath(
+      e,
+      "`test`.`not_existing_table`",
+      searchPath = "not available")
+  }
+
   test("create a table") {
     withTable("h2.test.new_table") {
       sql("CREATE TABLE h2.test.new_table(i INT, j STRING)")

sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/state/StateStoreErrorsSuite.scala

@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+
+package org.apache.spark.sql.execution.streaming.state
+
+import org.apache.spark.SparkFunSuite
+
+class StateStoreErrorsSuite extends SparkFunSuite {
+
+  test("SPARK-58945: state store mismatch reports schema details") {
+    checkError(
+      exception = StateStoreErrors.stateStoreColumnFamilyMismatch(
+        "state", "old_schema", "new_schema"),
+      condition = "STATE_STORE_COLUMN_FAMILY_SCHEMA_INCOMPATIBLE",
+      parameters = Map(
+        "colFamilyName" -> "state",
+        "oldSchema" -> "old_schema",
+        "newSchema" -> "new_schema"))
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/scripting/SqlScriptingCursorE2eSuite.scala

@@ -117,6 +117,15 @@ class SqlScriptingCursorE2eSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-58945: OPEN cursor outside script reports cursor name") {
+    checkError(
+      exception = intercept[AnalysisException] {
+        sql("OPEN cur")
+      },
+      condition = "CURSOR_OUTSIDE_SCRIPT",
+      parameters = Map("cursorName" -> "`cur`"))
+  }
+
   test("Test 5: Cursors have a separate namespace from local variables") {
     val result = sql(
       """