diff --git a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala b/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala index 244d79aa217a4..e60ed8164b862 100644 --- a/core/src/main/scala/org/apache/spark/rdd/ReliableCheckpointRDD.scala +++ b/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. */ diff --git a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala b/core/src/test/scala/org/apache/spark/CheckpointSuite.scala index af06e8f29f3b7..07cc15dd931c7 100644 --- a/core/src/test/scala/org/apache/spark/CheckpointSuite.scala +++ b/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