From e6695ed92fd899eb199443debdeb0093967e4425 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 6 Oct 2026 09:14:05 -0700 Subject: [PATCH 1/2] [Spark] REPLACE PARTITIONED BY: rewrite data files that do not store partition columns When a partitioned table was written without the partition column values in the Parquet files (the default), ALTER TABLE ... REPLACE PARTITIONED BY now rewrites only those files within the same commit, instead of failing. Files that already store the partition columns are kept as-is (metadata-only). The rewrite mirrors OPTIMIZE: deletion vectors are applied, row tracking columns are preserved, and all actions are dataChange=false. Controlled by spark.databricks.delta.alterTable.replacePartitionedBy.rewriteNonMaterializedFiles (default true); when disabled, the command fails with an error listing the files and the config. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../resources/error/delta-error-classes.json | 2 +- .../apache/spark/sql/delta/DeltaErrors.scala | 5 +- ...rtitionedByWithClusterByDeltaCommand.scala | 113 ++++++++++++------ .../sql/delta/sources/DeltaSQLConf.scala | 10 ++ ...placePartitionedByWithClusterBySuite.scala | 97 ++++++++++++++- 5 files changed, 186 insertions(+), 41 deletions(-) diff --git a/spark/src/main/resources/error/delta-error-classes.json b/spark/src/main/resources/error/delta-error-classes.json index df1f057d1fa..bcaf3addbc1 100644 --- a/spark/src/main/resources/error/delta-error-classes.json +++ b/spark/src/main/resources/error/delta-error-classes.json @@ -2763,7 +2763,7 @@ "DELTA_REPLACE_PARTITIONED_BY_PARTITION_COLUMNS_NOT_MATERIALIZED" : { "message" : [ "Cannot replace the partitioning of the table because the partition column(s) are not stored in data file(s), for example: .", - "Rewrite these files so that they contain the partition column values before running ALTER TABLE REPLACE PARTITIONED BY." + "Enable to rewrite these files as part of the command, or rewrite them so that they contain the partition column values before running ALTER TABLE REPLACE PARTITIONED BY." ], "sqlState" : "55000" }, diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala index 572db6ca242..7ca0a9648e4 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala @@ -4284,7 +4284,10 @@ trait DeltaErrorsBase new DeltaAnalysisException( errorClass = "DELTA_REPLACE_PARTITIONED_BY_PARTITION_COLUMNS_NOT_MATERIALIZED", messageParameters = Array( - partitionColumns.mkString(", "), numFiles.toString, examplePaths.mkString(", "))) + partitionColumns.mkString(", "), + numFiles.toString, + examplePaths.mkString(", "), + DeltaSQLConf.DELTA_REPLACE_PARTITIONED_BY_REWRITE_NON_MATERIALIZED_FILES.key)) } def createTableWithDifferentClusteringException( diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/AlterTableReplacePartitionedByWithClusterByDeltaCommand.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/AlterTableReplacePartitionedByWithClusterByDeltaCommand.scala index 00fc22335ff..696741d7de6 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/commands/AlterTableReplacePartitionedByWithClusterByDeltaCommand.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/commands/AlterTableReplacePartitionedByWithClusterByDeltaCommand.scala @@ -24,7 +24,7 @@ import scala.util.Try import org.apache.spark.sql.delta.skipping.clustering.ClusteredTableUtils import org.apache.spark.sql.delta.skipping.clustering.temp.ClusterBySpec import org.apache.spark.sql.delta._ -import org.apache.spark.sql.delta.actions.Action +import org.apache.spark.sql.delta.actions.{Action, AddFile, FileAction} import org.apache.spark.sql.delta.catalog.DeltaTableV2 import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.stats.DeltaStatistics @@ -41,23 +41,27 @@ import org.apache.spark.sql.types._ import org.apache.spark.util.SerializableConfiguration /** - * Command that replaces the partitioning of a Delta table with clustering, without rewriting - * any data file: + * Command that replaces the partitioning of a Delta table with clustering: * - ALTER TABLE .. REPLACE PARTITIONED BY WITH CLUSTER BY (col1, col2, ...) * - ALTER TABLE .. REPLACE PARTITIONED BY WITH CLUSTER BY NONE * - * The conversion is a single metadata-only commit which - * 1. clears the partition columns in the table metadata, - * 2. re-adds every active file with empty `partitionValues` and `dataChange = false`, keeping its - * path, deletion vector and row tracking fields. Per-file min/max/nullCount statistics are - * synthesized for the former partition columns from the (constant) partition values so that - * data skipping on these columns keeps working, and - * 3. when clustering columns are given, enables clustering and records the clustering columns. - * * After the conversion the former partition columns are read from the data files, so they must be * physically stored in every data file (see `delta.writePartitionColumnsToParquet` and the - * `materializePartitionColumns` table feature). This is verified by reading the Parquet footers - * before committing. + * `materializePartitionColumns` table feature). The command reads the Parquet footers of all data + * files to find the files that do not store them. + * + * The conversion is a single commit which + * 1. clears the partition columns in the table metadata, + * 2. re-adds every file that stores the partition columns with empty `partitionValues` and + * `dataChange = false`, keeping its path, deletion vector and row tracking fields. Per-file + * min/max/nullCount statistics are synthesized for the former partition columns from the + * (constant) partition values so that data skipping on these columns keeps working, + * 3. rewrites every file that does not store the partition columns (if any), like OPTIMIZE does: + * the new files store all columns, deletion vectors are applied and row IDs and row commit + * versions are preserved, and + * 4. when clustering columns are given, enables clustering and records the clustering columns. + * + * If all files store the partition columns, the conversion does not rewrite any data. * * The commit is done with [[OptimisticTransaction.commitLarge]], which fails if any other commit * happened since the transaction started. Hence no file can be added concurrently between the @@ -80,11 +84,20 @@ case class AlterTableReplacePartitionedByWithClusterByDeltaCommand( ClusteredTableUtils.validateNumClusteringColumns(clusteringColumns, Some(deltaLog)) val physicalPartitionSchema = oldMetadata.physicalPartitionSchema - if (sparkSession.sessionState.conf.getConf( + val conf = sparkSession.sessionState.conf + val filesToRewrite = if (conf.getConf( DeltaSQLConf.DELTA_REPLACE_PARTITIONED_BY_VERIFY_MATERIALIZED_PARTITION_COLUMNS)) { - ReplacePartitionedByUtils.verifyPartitionColumnsMaterialized( - sparkSession, deltaLog, snapshot, oldMetadata.partitionColumns, - physicalPartitionSchema.fieldNames.toSeq) + ReplacePartitionedByUtils.findFilesWithoutPartitionColumns( + sparkSession, deltaLog, snapshot, physicalPartitionSchema.fieldNames.toSeq) + } else { + Seq.empty + } + if (filesToRewrite.nonEmpty && !conf.getConf( + DeltaSQLConf.DELTA_REPLACE_PARTITIONED_BY_REWRITE_NON_MATERIALIZED_FILES)) { + throw DeltaErrors.replacePartitionedByPartitionColumnsNotMaterializedException( + oldMetadata.partitionColumns, + filesToRewrite.size, + filesToRewrite.take(ReplacePartitionedByUtils.NUM_EXAMPLE_FILES).map(_.path)) } val newLogicalClusteringColumns = clusteringColumns.map(FieldReference(_).toString) @@ -106,12 +119,17 @@ case class AlterTableReplacePartitionedByWithClusterByDeltaCommand( Nil } - val stringPrefixLength = sparkSession.sessionState.conf.getConf( - DeltaSQLConf.DATA_SKIPPING_STRING_PREFIX_LENGTH) + // Must run after the metadata update, so that the new files store all columns. + val rewriteActions = + ReplacePartitionedByUtils.rewriteFiles(sparkSession, txn, snapshot, filesToRewrite) + + val stringPrefixLength = conf.getConf(DeltaSQLConf.DATA_SKIPPING_STRING_PREFIX_LENGTH) val partitionFields = physicalPartitionSchema.fields.toSeq + val rewrittenPaths = sparkSession.sparkContext.broadcast(filesToRewrite.map(_.path).toSet) import org.apache.spark.sql.delta.implicits._ val newAddFiles = snapshot.allFiles.mapPartitions { files => - files.map { file => + val skip = rewrittenPaths.value + files.filterNot(file => skip.contains(file.path)).map { file => file.copy( partitionValues = Map.empty, dataChange = false, @@ -121,7 +139,12 @@ case class AlterTableReplacePartitionedByWithClusterByDeltaCommand( } val actions: Iterator[Action] = - newAddFiles.toLocalIterator().asScala ++ domainMetadata.iterator + rewriteActions.iterator ++ newAddFiles.toLocalIterator().asScala ++ domainMetadata.iterator + val metrics = Map( + "numRewrittenFiles" -> filesToRewrite.size.toString, + "numRewrittenBytes" -> filesToRewrite.map(_.size).sum.toString, + "numAddedFilesFromRewrite" -> + rewriteActions.count(_.isInstanceOf[AddFile]).toString) val newProtocolOpt = if (txn.protocol != snapshot.protocol) Some(txn.protocol) else None txn.commitLarge( @@ -132,7 +155,7 @@ case class AlterTableReplacePartitionedByWithClusterByDeltaCommand( oldMetadata.partitionColumns.mkString(","), newLogicalClusteringColumns.mkString(",")), context = Map.empty, - metrics = Map.empty, + metrics = metrics, dataChange = Some(false)) } Seq.empty[Row] @@ -141,28 +164,27 @@ case class AlterTableReplacePartitionedByWithClusterByDeltaCommand( object ReplacePartitionedByUtils { - private val NUM_EXAMPLE_FILES = 5 + val NUM_EXAMPLE_FILES = 5 /** - * Verifies that every active data file of the snapshot physically stores all the given + * Returns the active data files of the snapshot that do not physically store all the given * partition columns, by reading the Parquet footers of the files. */ - def verifyPartitionColumnsMaterialized( + def findFilesWithoutPartitionColumns( spark: SparkSession, deltaLog: DeltaLog, snapshot: Snapshot, - logicalPartitionColumns: Seq[String], - physicalPartitionColumns: Seq[String]): Unit = { + physicalPartitionColumns: Seq[String]): Seq[AddFile] = { val broadcastConf = spark.sparkContext.broadcast( new SerializableConfiguration(deltaLog.newDeltaHadoopConf())) val dataRootDir = deltaLog.dataPath.toString val requiredColumns = physicalPartitionColumns.map(_.toLowerCase(Locale.ROOT)) import org.apache.spark.sql.delta.implicits._ - val filesMissingColumns = snapshot.allFiles.select("path").as[String].mapPartitions { paths => + snapshot.allFiles.mapPartitions { files => val conf = broadcastConf.value.value - paths.filter { relativePath => - val path = DeltaFileOperations.absolutePath(dataRootDir, relativePath) + files.filter { file => + val path = DeltaFileOperations.absolutePath(dataRootDir, file.path) val status = path.getFileSystem(conf).getFileStatus(path) val footer = ParquetFooterReaderShims.readParquetFooter( conf, status, ParquetMetadataConverter.SKIP_ROW_GROUPS) @@ -170,13 +192,34 @@ object ReplacePartitionedByUtils { .map(_.getName.toLowerCase(Locale.ROOT)).toSet !requiredColumns.forall(fileColumns.contains) } - } + }.collect().toSeq + } - val examples = filesMissingColumns.take(NUM_EXAMPLE_FILES) - if (examples.nonEmpty) { - throw DeltaErrors.replacePartitionedByPartitionColumnsNotMaterializedException( - logicalPartitionColumns, filesMissingColumns.count(), examples.toSeq) + /** + * Rewrites the given files with the metadata of the transaction, which must not have partition + * columns anymore, so that the new files store all columns. Deletion vectors are applied, and + * row IDs and row commit versions are preserved like in OPTIMIZE. Returns the new files and the + * removal of the given files, all with `dataChange = false`. + */ + def rewriteFiles( + spark: SparkSession, + txn: OptimisticTransaction, + snapshot: Snapshot, + files: Seq[AddFile]): Seq[FileAction] = { + if (files.isEmpty) return Seq.empty + require(txn.metadata.partitionColumns.isEmpty, + "The partitioning must be dropped before rewriting the files") + var input = txn.deltaLog.createDataFrame( + snapshot, files, actionTypeOpt = Some("ReplacePartitionedBy")) + input = RowTracking.preserveRowTrackingColumns(input, snapshot) + val addFiles = txn.writeFiles(input, None, isOptimize = true, Nil).collect { + case a: AddFile => a.copy(dataChange = false) + case other => + throw new IllegalStateException( + s"Unexpected action $other with type ${other.getClass} while rewriting files") } + val timestamp = System.currentTimeMillis() + addFiles ++ files.map(_.removeWithTimestamp(timestamp, dataChange = false)) } /** diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala index ceed56b9cc3..45faa6381e3 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala @@ -3455,6 +3455,16 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .booleanConf .createWithDefault(true) + val DELTA_REPLACE_PARTITIONED_BY_REWRITE_NON_MATERIALIZED_FILES = + buildConf("alterTable.replacePartitionedBy.rewriteNonMaterializedFiles") + .internal() + .doc("""When true, ALTER TABLE ... REPLACE PARTITIONED BY WITH CLUSTER BY rewrites the data + |files that do not physically store the partition columns, so that the partition column + |values are stored in the new files. When false, the command fails if such files exist. + |""".stripMargin) + .booleanConf + .createWithDefault(true) + val DELTA_LOG_CACHE_SIZE = buildConf("delta.log.cacheSize") .internal() .doc("The maximum number of DeltaLog instances to cache in memory.") diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala index 6113ff27460..c55c617a062 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala @@ -19,7 +19,7 @@ package org.apache.spark.sql.delta.skipping.clustering import org.apache.spark.sql.delta.skipping.ClusteredTableTestUtils import org.apache.spark.sql.delta.skipping.clustering.temp.{AlterTableReplacePartitionedByWithClusterBy, ClusterBySpec} import org.apache.spark.sql.delta.{DeltaAnalysisException, DeltaLog} -import org.apache.spark.sql.delta.actions.AddFile +import org.apache.spark.sql.delta.actions.{AddFile, RemoveFile} import org.apache.spark.sql.delta.commands.ReplacePartitionedByUtils import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.stats.DataSkippingDeltaTestsUtils @@ -92,6 +92,8 @@ class ReplacePartitionedByWithClusterBySuite val changes = deltaLog.getChanges(after.version).next()._2 assert(changes.collect { case a: AddFile => a }.forall(!_.dataChange)) + assert(!changes.exists(_.isInstanceOf[RemoveFile])) + assert(lastCommit.operationMetrics.flatMap(_.get("numRewrittenFiles")) === Some("0")) // Time travel to the partitioned version still works. checkAnswer( @@ -245,7 +247,89 @@ class ReplacePartitionedByWithClusterBySuite } } - test("fails when data files do not store the partition columns") { + private def filesWithoutPartitionColumns(path: String, partitionColumns: String*) = { + val deltaLog = DeltaLog.forTable(spark, path) + ReplacePartitionedByUtils.findFilesWithoutPartitionColumns( + spark, deltaLog, deltaLog.update(), partitionColumns) + } + + private def lastOperationMetrics(path: String): Map[String, String] = + sql(s"DESCRIBE HISTORY delta.`$path` LIMIT 1") + .select("operationMetrics").head().getMap[String, String](0).toMap + + test("rewrites the data files that do not store the partition columns") { + withTempDir { dir => + val path = dir.getCanonicalPath + sql(s"""CREATE TABLE delta.`$path` (id BIGINT, p STRING) USING delta PARTITIONED BY (p) + |TBLPROPERTIES ('delta.writePartitionColumnsToParquet' = 'false')""".stripMargin) + sql(s"INSERT INTO delta.`$path` VALUES (1, 'a'), (2, 'b')") + sql(s"ALTER TABLE delta.`$path` SET TBLPROPERTIES " + + "('delta.writePartitionColumnsToParquet' = 'true')") + sql(s"INSERT INTO delta.`$path` VALUES (3, 'a'), (4, null)") + val deltaLog = DeltaLog.forTable(spark, path) + val before = deltaLog.update() + val notMaterialized = filesWithoutPartitionColumns(path, "p").map(_.path).toSet + assert(notMaterialized.size === 2) + val materialized = before.allFiles.collect().map(_.path).toSet -- notMaterialized + assert(materialized.size === 2) + + sql(s"ALTER TABLE delta.`$path` REPLACE PARTITIONED BY WITH CLUSTER BY NONE") + + val after = deltaLog.update() + assert(after.version === before.version + 1) + assert(after.metadata.partitionColumns.isEmpty) + val afterFiles = after.allFiles.collect() + assert(afterFiles.forall(_.partitionValues.isEmpty)) + val afterPaths = afterFiles.map(_.path).toSet + assert(materialized.subsetOf(afterPaths)) + assert((afterPaths intersect notMaterialized).isEmpty) + assert(filesWithoutPartitionColumns(path, "p").isEmpty) + + val changes = deltaLog.getChanges(after.version).next()._2 + val removed = changes.collect { case r: RemoveFile => r } + assert(removed.map(_.path).toSet === notMaterialized) + assert(removed.forall(!_.dataChange)) + assert(changes.collect { case a: AddFile => a }.forall(!_.dataChange)) + val metrics = lastOperationMetrics(path) + assert(metrics("numRewrittenFiles") === "2") + assert(metrics("numAddedFilesFromRewrite").toInt > 0) + + checkAnswer(readTable(path), + Seq(Row(1L, "a"), Row(2L, "b"), Row(3L, "a"), Row(4L, null))) + checkAnswer(readTable(path).where("p = 'b'").select("id"), Row(2L)) + checkAnswer(readTable(path).where("p IS NULL").select("id"), Row(4L)) + } + } + + test("rewrite applies deletion vectors and preserves row tracking") { + withTempDir { dir => + val path = dir.getCanonicalPath + sql(s"""CREATE TABLE delta.`$path` (id BIGINT, p STRING) USING delta PARTITIONED BY (p) + |TBLPROPERTIES ( + | 'delta.writePartitionColumnsToParquet' = 'false', + | 'delta.enableDeletionVectors' = 'true', + | 'delta.enableRowTracking' = 'true')""".stripMargin) + spark.range(0, 100, 1, 4).selectExpr("id", "cast(id % 3 as string) as p") + .write.format("delta").mode("append").save(path) + sql(s"DELETE FROM delta.`$path` WHERE id % 10 = 0") + val deltaLog = DeltaLog.forTable(spark, path) + assert(deltaLog.update().allFiles.collect().exists(_.deletionVector != null)) + val rowTrackingColumns = + Seq("*", "_metadata.row_id", "_metadata.row_commit_version") + val expected = readTable(path).selectExpr(rowTrackingColumns: _*).collect() + assert(expected.length === 90) + + sql(s"ALTER TABLE delta.`$path` REPLACE PARTITIONED BY WITH CLUSTER BY NONE") + + val after = deltaLog.update() + assert(after.metadata.partitionColumns.isEmpty) + assert(after.allFiles.collect().forall(_.deletionVector == null)) + assert(filesWithoutPartitionColumns(path, "p").isEmpty) + checkAnswer(readTable(path).selectExpr(rowTrackingColumns: _*), expected) + } + } + + test("fails when data files do not store the partition columns and rewrite is disabled") { withTempDir { dir => val path = dir.getCanonicalPath sql(s"""CREATE TABLE delta.`$path` (id BIGINT, p STRING) USING delta PARTITIONED BY (p) @@ -253,12 +337,17 @@ class ReplacePartitionedByWithClusterBySuite sql(s"INSERT INTO delta.`$path` VALUES (1, 'a'), (2, 'b')") val deltaLog = DeltaLog.forTable(spark, path) val versionBefore = deltaLog.update().version - val e = intercept[DeltaAnalysisException] { - sql(s"ALTER TABLE delta.`$path` REPLACE PARTITIONED BY WITH CLUSTER BY NONE") + val e = withSQLConf(DeltaSQLConf + .DELTA_REPLACE_PARTITIONED_BY_REWRITE_NON_MATERIALIZED_FILES.key -> "false") { + intercept[DeltaAnalysisException] { + sql(s"ALTER TABLE delta.`$path` REPLACE PARTITIONED BY WITH CLUSTER BY NONE") + } } assert(e.getErrorClass === "DELTA_REPLACE_PARTITIONED_BY_PARTITION_COLUMNS_NOT_MATERIALIZED") assert(e.getMessage.contains("2 data file(s)")) + assert(e.getMessage.contains( + DeltaSQLConf.DELTA_REPLACE_PARTITIONED_BY_REWRITE_NON_MATERIALIZED_FILES.key)) assert(deltaLog.update().version === versionBefore) assert(deltaLog.update().metadata.partitionColumns === Seq("p")) } From 0351c6e216993789c839b241fa010617d94a6bd9 Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 6 Oct 2026 10:13:33 -0700 Subject: [PATCH 2/2] Test stats of rewritten files for the former partition columns Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../ReplacePartitionedByWithClusterBySuite.scala | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala index c55c617a062..9efaec72d9b 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/skipping/clustering/ReplacePartitionedByWithClusterBySuite.scala @@ -284,6 +284,18 @@ class ReplacePartitionedByWithClusterBySuite assert(materialized.subsetOf(afterPaths)) assert((afterPaths intersect notMaterialized).isEmpty) assert(filesWithoutPartitionColumns(path, "p").isEmpty) + // Synthesized stats for kept files, collected stats for rewritten files. + afterFiles.foreach { f => + assert(statsOf(f)("nullCount").asInstanceOf[Map[String, Any]].contains("p")) + } + val rewrittenStats = afterFiles.filterNot(f => materialized.contains(f.path)).map(statsOf) + assert(rewrittenStats.nonEmpty) + def pStat(kind: String): Seq[String] = + rewrittenStats.map(_(kind).asInstanceOf[Map[String, Any]]("p").toString).toSeq + assert(pStat("minValues").min === "a") + assert(pStat("maxValues").max === "b") + assert(filesRead(spark, deltaLog, "p = 'b'", checkEmptyUnusedFilters = false) < + afterFiles.length) val changes = deltaLog.getChanges(after.version).next()._2 val removed = changes.collect { case r: RemoveFile => r }