[SPARK-58881][CORE] Ignore non-numeric checkpoint part files

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

@@ -74,7 +74,7 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
     // listStatus can throw exception if path does not exist.
     val inputFiles = fs.listStatus(cpath)
       .map(_.getPath)
-      .filter(_.getName.startsWith("part-"))
+      .filter(path => ReliableCheckpointRDD.isCheckpointFile(path.getName))
       .sortBy(_.getName.stripPrefix("part-").toInt)
     // Fail fast if input files are invalid
     inputFiles.zipWithIndex.foreach { case (path, i) =>
@@ -132,6 +132,12 @@ private[spark] class ReliableCheckpointRDD[T: ClassTag](
 
 private[spark] object ReliableCheckpointRDD extends Logging {
 
+  private def isCheckpointFile(fileName: String): Boolean = {
+    val partitionId = fileName.stripPrefix("part-")
+    fileName.startsWith("part-") && partitionId.nonEmpty &&
+      partitionId.forall(c => c >= '0' && c <= '9')
+  }
+
   /**
    * Return the checkpoint file name for the given partition.
    */

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

@@ -750,6 +750,26 @@ class CheckpointStorageSuite extends SparkFunSuite with LocalSparkContext {
     }
   }
 
+  test("SPARK-58881: ignore non-numeric part files in a checkpoint directory") {
+    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))
+
+      val checkpointPath = new Path(rdd.getCheckpointFile.get)
+      val fs = checkpointPath.getFileSystem(sc.hadoopConfiguration)
+      Seq("part-00000.bak", "part-backup", "part-").foreach { fileName =>
+        fs.create(new Path(checkpointPath, fileName)).close()
+      }
+
+      val recovered = sc.checkpointFile[Int](checkpointPath.toString)
+      assert(recovered.getNumPartitions === 4)
+      assert(recovered.collect().toSeq === (1 to 20))
+    }
+  }
+
   test("checkpoint path that cannot be created") {
     withTempDir { checkpointDir =>
       // MkdirsFailingFilesystem refuses to create the per-RDD directory and reports it the way