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

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

@@ -11723,7 +11723,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/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -4675,7 +4675,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

@@ -3215,8 +3215,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 = {
@@ -3240,7 +3239,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/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

@@ -228,7 +228,12 @@ private[sql] case class H2Dialect() extends JdbcDialect with NoLegacyJDBCError {
             val relationName = messageParameters.getOrElse("tableName", "")
             throw new NoSuchTableException(
               errorClass = "TABLE_OR_VIEW_NOT_FOUND",
-              messageParameters = Map("relationName" -> relationName),
+              messageParameters = Map(
+                "relationName" -> relationName,
+                // classifyException receives only pre-rendered strings, so no Spark-side
+                // search path exists here; "not available" matches the rendering used for
+                // an empty search path.
+                "searchPath" -> "not available"),
               cause = Some(e))
           // SCHEMA_NOT_FOUND_1
           case 90079 =>

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

@@ -0,0 +1,52 @@
+/*
+ * 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.errors
+
+import org.apache.spark.{SparkFunSuite, SparkRuntimeException}
+import org.apache.spark.sql.catalyst.catalog.InvalidUDFClassException
+import org.apache.spark.sql.execution.streaming.state.StateStoreColumnFamilyMismatch
+
+class ErrorMessageParametersSuite extends SparkFunSuite {
+
+  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("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("SPARK-58945: state store mismatch reports schema details") {
+    checkError(
+      exception = new 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/errors/QueryExecutionErrorsSuite.scala

@@ -141,6 +141,21 @@ class QueryExecutionErrorsSuite
     (df1, df2)
   }
 
+  test("SPARK-58945: invalid file extension reports invalidValue") {
+    withTempDir { dir =>
+      val path = new File(dir, "data").getCanonicalPath
+      checkError(
+        exception = intercept[SparkIllegalArgumentException] {
+          spark.range(1).write.option("extension", "12").csv(path)
+        },
+        condition = "INVALID_PARAMETER_VALUE.EXTENSION",
+        parameters = Map(
+          "functionName" -> "`extension`",
+          "parameter" -> "`extension`",
+          "invalidValue" -> "`12`"))
+    }
+  }
+
   test("INVALID_PARAMETER_VALUE.AES_KEY_LENGTH: invalid key lengths in AES functions") {
     val (df1, df2) = getAesInputs()
     def checkInvalidKeyLength(df: => DataFrame, inputBytes: Int): Unit = {

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

@@ -16,7 +16,7 @@
  */
 package org.apache.spark.sql.execution.datasources.v2.jdbc
 
-import java.sql.{Connection, DriverManager}
+import java.sql.{Connection, DriverManager, SQLException}
 import java.util.Properties
 
 import scala.jdk.CollectionConverters._
@@ -25,13 +25,14 @@ 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}
 import org.apache.spark.sql.errors.DataTypeErrors.{toSQLConf, toSQLStmt}
 import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
 import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.jdbc.H2Dialect
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -229,6 +230,21 @@ class JDBCTableCatalogSuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-58945: H2 dialect supplies searchPath for TABLE_OR_VIEW_NOT_FOUND") {
+    val e = intercept[NoSuchTableException] {
+      H2Dialect().classifyException(
+        new SQLException("""Table "NOT_EXISTING_TABLE" not found""", "42S02", 42102),
+        condition = "FAILED_JDBC.LOAD_TABLE",
+        messageParameters = Map("tableName" -> "`test`.`not_existing_table`"),
+        description = "Failed to load table: [test, not_existing_table]",
+        isRuntime = false)
+    }
+    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)")