[SPARK-40400][SQL] Pass error message parameters to exceptions as maps

core/src/main/java/org/apache/spark/memory/MemoryConsumer.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 
+import org.apache.spark.errors.SparkCoreErrors;
 import org.apache.spark.unsafe.array.LongArray;
 import org.apache.spark.unsafe.memory.MemoryBlock;
 
@@ -153,9 +154,6 @@ private void throwOom(final MemoryBlock page, final long required) {
       taskMemoryManager.freePage(page, this);
     }
     taskMemoryManager.showMemoryUsage();
-    // checkstyle.off: RegexpSinglelineJava
-    throw new SparkOutOfMemoryError("UNABLE_TO_ACQUIRE_MEMORY",
-            new String[]{Long.toString(required), Long.toString(got)});
-    // checkstyle.on: RegexpSinglelineJava
+    throw SparkCoreErrors.outOfMemoryError(required, got);
   }
 }

core/src/main/java/org/apache/spark/memory/SparkOutOfMemoryError.java

@@ -20,6 +20,8 @@
 import org.apache.spark.SparkThrowableHelper;
 import org.apache.spark.annotation.Private;
 
+import java.util.Map;
+
 /**
  * This exception is thrown when a task can not acquire memory from the Memory manager.
  * Instead of throwing {@link OutOfMemoryError}, which kills the executor,
@@ -28,7 +30,7 @@
 @Private
 public final class SparkOutOfMemoryError extends OutOfMemoryError implements SparkThrowable {
     String errorClass;
-    String[] messageParameters;
+    Map<String, String> messageParameters;
 
     public SparkOutOfMemoryError(String s) {
         super(s);
@@ -38,15 +40,15 @@ public SparkOutOfMemoryError(OutOfMemoryError e) {
         super(e.getMessage());
     }
 
-    public SparkOutOfMemoryError(String errorClass, String[] messageParameters) {
+    public SparkOutOfMemoryError(String errorClass, Map<String, String> messageParameters) {
         super(SparkThrowableHelper.getMessage(errorClass, null, messageParameters));
         this.errorClass = errorClass;
         this.messageParameters = messageParameters;
     }
 
     @Override
     public String[] getMessageParameters() {
-        return messageParameters;
+        return SparkThrowableHelper.getMessageParameters(errorClass, null, messageParameters);
     }
 
     @Override

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

@@ -484,7 +484,7 @@
   },
   "UNRESOLVED_MAP_KEY" : {
     "message" : [
-      "Cannot resolve column <columnName> as a map key. If the key is a string literal, please add single quotes around it."
+      "Cannot resolve column <objectName> as a map key. If the key is a string literal, please add single quotes around it."
     ],
     "subClass" : {
       "WITHOUT_SUGGESTION" : {