[SPARK-38477][CORE] Use error class in org.apache.spark.shuffle

### What changes were proposed in this pull request?
This PR aims to change exceptions created in package org.apache.spark.shuffle to use error class. This also adds an error class INTERNAL_ERROR_SHUFFLE and uses that for the internal errors in the package.

### Why are the changes needed?
This is to move exceptions created in package org.apache.spark.shuffle to error class.

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

### How was this patch tested?
Existing tests.

Closes #41575 from bozhang2820/spark-38477.

Authored-by: Bo Zhang <[email protected]>
Signed-off-by: Max Gekk <[email protected]>

Author: bozhang2820

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

@@ -729,6 +729,11 @@
     ],
     "sqlState" : "42K04"
   },
+  "FAILED_RENAME_TEMP_FILE" : {
+    "message" : [
+      "Failed to rename temp file <srcPath> to <dstPath> as FileSystem.rename returned false."
+    ]
+  },
   "FIELDS_ALREADY_EXISTS" : {
     "message" : [
       "Cannot <op> column, because <fieldNames> already exists in <struct>."
@@ -1024,6 +1029,12 @@
     ],
     "sqlState" : "XX000"
   },
+  "INTERNAL_ERROR_SHUFFLE" : {
+    "message" : [
+      "<message>"
+    ],
+    "sqlState" : "XX000"
+  },
   "INTERVAL_ARITHMETIC_OVERFLOW" : {
     "message" : [
       "<message>.<alternative>"
@@ -5300,11 +5311,6 @@
       "Failed to rename as <dstPath> already exists."
     ]
   },
-  "_LEGACY_ERROR_TEMP_2199" : {
-    "message" : [
-      "Failed to rename temp file <srcPath> to <dstPath> as rename returned false."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_2200" : {
     "message" : [
       "Error: we detected a possible problem with the location of your \"_spark_metadata\"",

core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.errors
 
-import java.io.IOException
+import java.io.{File, IOException}
 import java.util.concurrent.TimeoutException
 
 import scala.collection.JavaConverters._
@@ -467,6 +467,15 @@ private[spark] object SparkCoreErrors {
         "receivedBytes" -> receivedBytes.toString).asJava)
   }
 
+  def failedRenameTempFileError(srcFile: File, dstFile: File): Throwable = {
+    new SparkException(
+      errorClass = "FAILED_RENAME_TEMP_FILE",
+      messageParameters = Map(
+        "srcPath" -> srcFile.toString,
+        "dstPath" -> dstFile.toString),
+      cause = null)
+  }
+
   def addLocalDirectoryError(path: Path): Throwable = {
     new SparkException(
       errorClass = "UNSUPPORTED_ADD_FILE.LOCAL_DIRECTORY",

core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala

@@ -25,6 +25,7 @@ import java.nio.file.Files
 import scala.collection.mutable.ArrayBuffer
 
 import org.apache.spark.{SparkConf, SparkEnv, SparkException}
+import org.apache.spark.errors.SparkCoreErrors
 import org.apache.spark.internal.{config, Logging}
 import org.apache.spark.io.NioBufferedFileInputStream
 import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
@@ -227,7 +228,8 @@ private[spark] class IndexShuffleBlockResolver(
     remoteShuffleMaxDisk.foreach { maxBytes =>
       val bytesUsed = getShuffleBytesStored()
       if (maxBytes < bytesUsed) {
-        throw new SparkException(s"Not storing remote shuffles $bytesUsed exceeds $maxBytes")
+        throw SparkException.internalError(
+          s"Not storing remote shuffles $bytesUsed exceeds $maxBytes", category = "SHUFFLE")
       }
     }
     val file = blockId match {
@@ -236,8 +238,8 @@ private[spark] class IndexShuffleBlockResolver(
       case ShuffleDataBlockId(shuffleId, mapId, _) =>
         getDataFile(shuffleId, mapId)
       case _ =>
-        throw new IllegalStateException(s"Unexpected shuffle block transfer ${blockId} as " +
-          s"${blockId.getClass().getSimpleName()}")
+        throw SparkException.internalError(s"Unexpected shuffle block transfer $blockId as " +
+          s"${blockId.getClass().getSimpleName()}", category = "SHUFFLE")
     }
     val fileTmp = createTempFile(file)
     val channel = Channels.newChannel(
@@ -263,7 +265,7 @@ private[spark] class IndexShuffleBlockResolver(
             file.delete()
           }
           if (!fileTmp.renameTo(file)) {
-            throw new IOException(s"fail to rename file ${fileTmp} to ${file}")
+            throw SparkCoreErrors.failedRenameTempFileError(fileTmp, file)
           }
         }
         blockManager.reportBlockStatus(blockId, BlockStatus(StorageLevel.DISK_ONLY, 0, diskSize))
@@ -300,7 +302,7 @@ private[spark] class IndexShuffleBlockResolver(
 
       // Make sure the index exist.
       if (!indexFile.exists()) {
-        throw new FileNotFoundException("Index file is deleted already.")
+        throw SparkException.internalError("Index file is deleted already.", category = "SHUFFLE")
       }
       if (dataFile.exists()) {
         List((dataBlockId, dataBlockData), (indexBlockId, indexBlockData))
@@ -389,7 +391,7 @@ private[spark] class IndexShuffleBlockResolver(
             dataFile.delete()
           }
           if (dataTmp != null && dataTmp.exists() && !dataTmp.renameTo(dataFile)) {
-            throw new IOException("fail to rename file " + dataTmp + " to " + dataFile)
+            throw SparkCoreErrors.failedRenameTempFileError(dataTmp, dataFile)
           }
 
           // write the checksum file
@@ -462,11 +464,10 @@ private[spark] class IndexShuffleBlockResolver(
     }
 
     if (!tmpFile.renameTo(targetFile)) {
-      val errorMsg = s"fail to rename file $tmpFile to $targetFile"
       if (propagateError) {
-        throw new IOException(errorMsg)
+        throw SparkCoreErrors.failedRenameTempFileError(tmpFile, targetFile)
       } else {
-        logWarning(errorMsg)
+        logWarning(s"fail to rename file $tmpFile to $targetFile")
       }
     }
   }
@@ -567,7 +568,8 @@ private[spark] class IndexShuffleBlockResolver(
       case batchId: ShuffleBlockBatchId =>
         (batchId.shuffleId, batchId.mapId, batchId.startReduceId, batchId.endReduceId)
       case _ =>
-        throw new IllegalArgumentException("unexpected shuffle block id format: " + blockId)
+        throw SparkException.internalError(
+          s"unexpected shuffle block id format: $blockId", category = "SHUFFLE")
     }
     // The block is actually going to be a range of a single map output file for this map, so
     // find out the consolidated file, then the offset within that from our index
@@ -589,8 +591,9 @@ private[spark] class IndexShuffleBlockResolver(
       val actualPosition = channel.position()
       val expectedPosition = endReduceId * 8L + 8
       if (actualPosition != expectedPosition) {
-        throw new Exception(s"SPARK-22982: Incorrect channel position after index file reads: " +
-          s"expected $expectedPosition but actual position was $actualPosition.")
+        throw SparkException.internalError(s"SPARK-22982: Incorrect channel position after index" +
+          s" file reads: expected $expectedPosition but actual position was $actualPosition.",
+          category = "SHUFFLE")
       }
       new FileSegmentManagedBuffer(
         transportConf,

core/src/main/scala/org/apache/spark/shuffle/ShufflePartitionPairsWriter.scala

@@ -17,9 +17,10 @@
 
 package org.apache.spark.shuffle
 
-import java.io.{Closeable, IOException, OutputStream}
+import java.io.{Closeable, OutputStream}
 import java.util.zip.Checksum
 
+import org.apache.spark.SparkException
 import org.apache.spark.io.MutableCheckedOutputStream
 import org.apache.spark.serializer.{SerializationStream, SerializerInstance, SerializerManager}
 import org.apache.spark.shuffle.api.ShufflePartitionWriter
@@ -53,7 +54,7 @@ private[spark] class ShufflePartitionPairsWriter(
 
   override def write(key: Any, value: Any): Unit = {
     if (isClosed) {
-      throw new IOException("Partition pairs writer is already closed.")
+      throw SparkException.internalError("Partition pairs writer is already closed.", "SHUFFLE")
     }
     if (objOut == null) {
       open()

docs/sql-error-conditions.md

@@ -494,6 +494,12 @@ Failed parsing struct: `<raw>`.
 
 Failed to rename `<sourcePath>` to `<targetPath>` as destination already exists.
 
+### FAILED_RENAME_TEMP_FILE
+
+SQLSTATE: none assigned
+
+Failed to rename temp file `<srcPath>` to `<dstPath>` as FileSystem.rename returned false.
+
 ### FIELDS_ALREADY_EXISTS
 
 SQLSTATE: none assigned
@@ -722,6 +728,12 @@ For more details see [INSUFFICIENT_TABLE_PROPERTY](sql-error-conditions-insuffic
 
 `<message>`
 
+### INTERNAL_ERROR_SHUFFLE
+
+[SQLSTATE: XX000](sql-error-conditions-sqlstates.html#class-XX-internal-error)
+
+`<message>`
+
 ### INTERVAL_ARITHMETIC_OVERFLOW
 
 [SQLSTATE: 22015](sql-error-conditions-sqlstates.html#class-22-data-exception)

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

@@ -1970,7 +1970,7 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
 
   def failedRenameTempFileError(srcPath: Path, dstPath: Path): Throwable = {
     new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_2199",
+      errorClass = "FAILED_RENAME_TEMP_FILE",
       messageParameters = Map(
         "srcPath" -> srcPath.toString(),
         "dstPath" -> dstPath.toString()),