[SPARK-58770][CORE] Assign a name to the error condition _LEGACY_ERROR_TEMP_3016-3020

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

@@ -993,6 +993,13 @@
     },
     "sqlState" : "XX000"
   },
+  "CHECKPOINT_DIRECTORY_NOT_SET" : {
+    "message" : [
+      "Cannot checkpoint because no checkpoint directory is configured.",
+      "Set one with `SparkContext.setCheckpointDir` or the \"spark.checkpoint.dir\" configuration."
+    ],
+    "sqlState" : "55019"
+  },
   "CHECKPOINT_FILE_CHECKSUM_VERIFICATION_FAILED" : {
     "message" : [
       "Checksum verification failed, the file may be corrupted. File: <fileName>",
@@ -1008,6 +1015,13 @@
     ],
     "sqlState" : "56000"
   },
+  "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH" : {
+    "message" : [
+      "The checkpoint of RDD <originalRDDId> has <newRDDLength> partition(s), but the RDD itself has <originalRDDLength>. The checkpoint RDD is <newRDDId>.",
+      "This usually means the checkpoint directory is not on storage that both the driver and the executors can read and write, or that some checkpoint files did not survive until the checkpoint was read back."
+    ],
+    "sqlState" : "58030"
+  },
   "CHECK_CONSTRAINT_VIOLATION" : {
     "message" : [
       "CHECK constraint <constraintName> <expression> violated by row with values:",
@@ -2662,6 +2676,12 @@
     ],
     "sqlState" : "42822"
   },
+  "FAILED_CREATE_CHECKPOINT_DIRECTORY" : {
+    "message" : [
+      "Failed to create the checkpoint directory <path> as FileSystem.mkdirs returned false."
+    ],
+    "sqlState" : "58030"
+  },
   "FAILED_EXECUTE_UDF" : {
     "message" : [
       "User defined function (<functionName>: (<signature>) => <result>) failed due to: <reason>."
@@ -3758,6 +3778,12 @@
     },
     "sqlState" : "42K03"
   },
+  "INVALID_CHECKPOINT_DIRECTORY" : {
+    "message" : [
+      "Cannot read the checkpoint directory <path>: expected the partition file <expectedFileName> but found <fileName>. The partition files must be numbered contiguously from part-00000."
+    ],
+    "sqlState" : "58030"
+  },
   "INVALID_CLONE_SESSION_REQUEST" : {
     "message" : [
       "Invalid session clone request."
@@ -11528,33 +11554,6 @@
       "countByValueApprox() does not support arrays"
     ]
   },
-  "_LEGACY_ERROR_TEMP_3016" : {
-    "message" : [
-      "Checkpoint directory has not been set in the SparkContext"
-    ]
-  },
-  "_LEGACY_ERROR_TEMP_3017" : {
-    "message" : [
-      "Invalid checkpoint file: <path>"
-    ]
-  },
-  "_LEGACY_ERROR_TEMP_3018" : {
-    "message" : [
-      "Failed to create checkpoint path <checkpointDirPath>"
-    ]
-  },
-  "_LEGACY_ERROR_TEMP_3019" : {
-    "message" : [
-      "Checkpoint RDD has a different number of partitions from original RDD. Original",
-      "RDD [ID: <originalRDDId>, num of partitions: <originalRDDLength>];",
-      "Checkpoint RDD [ID: <newRDDId>, num of partitions: <newRDDLength>]."
-    ]
-  },
-  "_LEGACY_ERROR_TEMP_3020" : {
-    "message" : [
-      "Checkpoint dir must be specified."
-    ]
-  },
   "_LEGACY_ERROR_TEMP_3028" : {
     "message" : [
       "<errorMsg>"

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

@@ -157,22 +157,30 @@ private[spark] object SparkCoreErrors {
 
   def checkpointDirectoryHasNotBeenSetInSparkContextError(): Throwable = {
     new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_3016", messageParameters = Map.empty, cause = null
+      errorClass = "CHECKPOINT_DIRECTORY_NOT_SET",
+      messageParameters = Map.empty,
+      cause = null
     )
   }
 
-  def invalidCheckpointFileError(path: Path): Throwable = {
+  def invalidCheckpointDirectoryError(
+      partitionFilePath: Path,
+      expectedFileName: String): Throwable = {
     new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_3017",
-      messageParameters = Map("path" -> s"$path"),
+      errorClass = "INVALID_CHECKPOINT_DIRECTORY",
+      messageParameters = Map(
+        "path" -> s"${partitionFilePath.getParent}",
+        "expectedFileName" -> expectedFileName,
+        "fileName" -> partitionFilePath.getName
+      ),
       cause = null
     )
   }
 
   def failToCreateCheckpointPathError(checkpointDirPath: Path): Throwable = {
     new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_3018",
-      messageParameters = Map("checkpointDirPath" -> s"$checkpointDirPath"),
+      errorClass = "FAILED_CREATE_CHECKPOINT_DIRECTORY",
+      messageParameters = Map("path" -> s"$checkpointDirPath"),
       cause = null
     )
   }
@@ -183,7 +191,7 @@ private[spark] object SparkCoreErrors {
       newRDDId: Int,
       newRDDLength: Int): Throwable = {
     new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_3019",
+      errorClass = "CHECKPOINT_RDD_PARTITION_COUNT_MISMATCH",
       messageParameters = Map(
         "originalRDDId" -> s"$originalRDDId",
         "originalRDDLength" -> s"$originalRDDLength",
@@ -200,9 +208,8 @@ private[spark] object SparkCoreErrors {
   }
 
   def mustSpecifyCheckpointDirError(): Throwable = {
-    new SparkException(
-      errorClass = "_LEGACY_ERROR_TEMP_3020", messageParameters = Map.empty, cause = null
-    )
+    SparkException.internalError(
+      "SparkContext.checkpointDir is unset when creating ReliableRDDCheckpointData.")
   }
 
   def askStandaloneSchedulerToShutDownExecutorsError(e: Exception): Throwable = {

core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala

@@ -78,8 +78,9 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
       .sortBy(_.getName.stripPrefix("part-").toInt)
     // Fail fast if input files are invalid
     inputFiles.zipWithIndex.foreach { case (path, i) =>
-      if (path.getName != ReliableCheckpointRDD.checkpointFileName(i)) {
-        throw SparkCoreErrors.invalidCheckpointFileError(path)
+      val expectedFileName = ReliableCheckpointRDD.checkpointFileName(i)
+      if (path.getName != expectedFileName) {
+        throw SparkCoreErrors.invalidCheckpointDirectoryError(path, expectedFileName)
       }
     }
     Array.tabulate(inputFiles.length)(i => new CheckpointRDDPartition(i))

core/src/test/scala/org/apache/spark/CheckpointSuite.scala

@@ -21,7 +21,7 @@ import java.io.File
 
 import scala.reflect.ClassTag
 
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{LocalFileSystem, Path}
 
 import org.apache.spark.internal.config.CACHE_CHECKPOINT_PREFERRED_LOCS_EXPIRE_TIME
 import org.apache.spark.internal.config.UI._
@@ -684,4 +684,71 @@ class CheckpointStorageSuite extends SparkFunSuite with LocalSparkContext {
       assert(flatMappedRDD.collect() === result)
     }
   }
+
+  test("checkpoint() without a checkpoint directory") {
+    sc = new SparkContext("local", "test", new SparkConf().set(UI_ENABLED.key, "false"))
+    checkError(
+      exception = intercept[SparkException](sc.makeRDD(1 to 4).checkpoint()),
+      condition = "CHECKPOINT_DIRECTORY_NOT_SET",
+      sqlState = Some("55019"))
+  }
+
+  test("reading a checkpoint directory with a missing partition file") {
+    withTempDir { checkpointDir =>
+      sc = new SparkContext("local", "test", new SparkConf().set(UI_ENABLED.key, "false"))
+      sc.setCheckpointDir(checkpointDir.toString)
+      val rdd = sc.makeRDD(1 to 20, numSlices = 4)
+      rdd.checkpoint()
+      assert(rdd.collect().toSeq === (1 to 20))
+
+      // Drop a middle partition file so the remaining names are no longer part-00000..part-0000N.
+      val checkpointPath = new Path(rdd.getCheckpointFile.get)
+      val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+      assert(fs.delete(new Path(checkpointPath, "part-00001"), false))
+
+      val recovered = sc.checkpointFile[Int](checkpointPath.toString)
+      checkError(
+        exception = intercept[SparkException](recovered.partitions),
+        condition = "INVALID_CHECKPOINT_DIRECTORY",
+        sqlState = Some("58030"),
+        parameters = Map(
+          "path" -> checkpointPath.toString,
+          "expectedFileName" -> "part-00001",
+          "fileName" -> "part-00002"))
+    }
+  }
+
+  test("checkpoint path that cannot be created") {
+    withTempDir { checkpointDir =>
+      // MkdirsFailingFilesystem refuses to create the per-RDD directory and reports it the way
+      // HDFS and S3A do, by returning false rather than throwing.
+      val conf = new SparkConf()
+        .set("spark.hadoop.fs.file.impl", classOf[MkdirsFailingFilesystem].getName)
+        .set("spark.hadoop.fs.file.impl.disable.cache", "true")
+        .set(UI_ENABLED.key, "false")
+      sc = new SparkContext("local", "test", conf)
+      sc.setCheckpointDir(checkpointDir.toString)
+      val rdd = sc.makeRDD(1 to 4)
+      val rddPath = ReliableRDDCheckpointData.checkpointPath(sc, rdd.id).get
+
+      rdd.checkpoint()
+      checkError(
+        exception = intercept[SparkException](rdd.collect()),
+        condition = "FAILED_CREATE_CHECKPOINT_DIRECTORY",
+        sqlState = Some("58030"),
+        parameters = Map("path" -> rddPath.toString))
+    }
+  }
+}
+
+/**
+ * A filesystem that refuses to create a per-RDD checkpoint directory, returning false from
+ * `mkdirs` instead of throwing. `LocalFileSystem` throws `FileAlreadyExistsException` when the
+ * path is taken, while HDFS and S3A can report the failure through the return value, which is
+ * what `ReliableCheckpointRDD` checks.
+ */
+class MkdirsFailingFilesystem extends LocalFileSystem {
+  override def mkdirs(f: Path): Boolean = {
+    if (f.getName.startsWith("rdd-")) false else super.mkdirs(f)
+  }
 }