diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala b/spark/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala index 07e4d5bda50..ec737383bbb 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala @@ -29,6 +29,7 @@ import scala.util.control.NonFatal import org.apache.spark.sql.delta._ import org.apache.spark.sql.delta.actions.{AddCDCFile, AddFile, FileAction, RemoveFile, SingleAction} import org.apache.spark.sql.delta.catalog.DeltaTableV2 +import org.apache.spark.sql.delta.commands.cdc.CDCReader import org.apache.spark.sql.delta.logging.DeltaLogKeys import org.apache.spark.sql.delta.sources.DeltaSQLConf import org.apache.spark.sql.delta.util.{DeltaCommitFileProvider, DeltaFileOperations, FileNames, JsonUtils, Utils => DeltaUtils} @@ -912,15 +913,57 @@ trait VacuumCommandImpl extends DeltaCommand { ((_: String) => false, (_: String) => false) } - // Use DeltaFileOperations.recursiveListDirs - val files = DeltaFileOperations.recursiveListDirs( + // The `_change_data` directory (CDF files) mirrors the table's partitioning and can hold a + // large fraction of the table's files. It is a first-level directory under the table root, so + // recursing it inline leaves a single task listing that entire subtree while the rest of the + // cluster is idle. When enabled, list it as a separate branch instead: its sub-directories are + // spread across tasks and its listing runs concurrently with the main table listing. The set + // of listed files is identical either way. + val changeDataPath = new Path(basePath, CDCReader.CDC_LOCATION) + val fs = changeDataPath.getFileSystem(hadoopConf.value.value) + val listChangeDataSeparately = applyHiddenFilters && + spark.sessionState.conf.getConf(DeltaSQLConf.DELTA_VACUUM_LIST_CHANGE_DATA_DIR_SEPARATELY) && + fs.exists(changeDataPath) + + // When listing `_change_data` separately, exclude it from the main tree so it is not listed + // twice. `hiddenDirNameFilter` receives the directory's name component, so matching + // CDC_LOCATION only excludes the top-level `_change_data` directory. + val mainDirFilter: String => Boolean = + if (listChangeDataSeparately) { + (name: String) => hiddenDirFilter(name) || name == CDCReader.CDC_LOCATION + } else { + hiddenDirFilter + } + + val mainListing = DeltaFileOperations.recursiveListDirs( spark, Seq(basePath), hadoopConf, - hiddenDirNameFilter = hiddenDirFilter, + hiddenDirNameFilter = mainDirFilter, hiddenFileNameFilter = hiddenFileFilter, fileListingParallelism = parallelism ) + + val listing = if (listChangeDataSeparately) { + // `recursiveListDirs` emits the *contents* of its roots, not the roots themselves, so emit + // the `_change_data` directory entry explicitly to preserve parity with the inline listing + // (empty-directory cleanup must be able to see it). + val changeDataDir = spark.createDataset( + Seq(SerializableFileStatus.fromStatus(fs.getFileStatus(changeDataPath)))) + val changeDataListing = DeltaFileOperations.recursiveListDirs( + spark, + Seq(changeDataPath.toString), + hadoopConf, + hiddenDirNameFilter = hiddenDirFilter, + hiddenFileNameFilter = hiddenFileFilter, + fileListingParallelism = parallelism + ) + mainListing.union(changeDataDir).union(changeDataListing) + } else { + mainListing + } + + val files = listing .map { f => // Make paths url-encoded (same pattern as VacuumCommand) val path = pathStringtoUrlEncodedString(f.path) 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 c2b62fc6976..c66f8f27c03 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 @@ -782,6 +782,20 @@ trait DeltaSQLConfBase extends DeltaSQLConfUtils { .checkValue(_ > 0, "parallelDelete.parallelism must be positive") .createOptional + val DELTA_VACUUM_LIST_CHANGE_DATA_DIR_SEPARATELY = + buildConf("vacuum.listing.changeDataDirSeparately.enabled") + .internal() + .doc("When true, VACUUM lists the table's `_change_data` directory as a separate listing " + + "branch instead of recursing into it inline as one of the table root's first-level " + + "directories. Because `_change_data` mirrors the table's partitioning (or is a single " + + "large flat directory), recursing it inline leaves one task listing the entire " + + "change-data subtree while the rest of the cluster is idle. Listing it separately " + + "spreads its sub-directories across tasks and lets its listing run concurrently with " + + "the main table listing. The set of files considered by VACUUM is unchanged; only the " + + "listing parallelism differs.") + .booleanConf + .createWithDefault(true) + val ENFORCE_DELETED_FILE_AND_LOG_RETENTION_DURATION_COMPATIBILITY = buildConf("vacuum.enforceDeletedFileAndLogRetentionDurationCompatibility") .internal() diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala index 6ce740dab21..f4eb0a56506 100644 --- a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala +++ b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaVacuumSuite.scala @@ -53,7 +53,7 @@ import org.apache.spark.sql.functions.{col, expr, lit} import org.apache.spark.sql.test.SharedSparkSession import org.apache.spark.sql.types._ import org.apache.spark.unsafe.types.UTF8String -import org.apache.spark.util.ManualClock +import org.apache.spark.util.{ManualClock, SerializableConfiguration} trait DeltaVacuumSuiteBase extends QueryTest with SharedSparkSession @@ -509,6 +509,94 @@ class DeltaVacuumSuite extends DeltaVacuumSuiteBase with DeltaSQLCommandTest { super.sparkConf.set("spark.sql.sources.parallelPartitionDiscovery.parallelism", "2") } + private def broadcastHadoopConf(): org.apache.spark.broadcast.Broadcast[ + SerializableConfiguration] = { + // scalastyle:off deltahadoopconfiguration + val conf = spark.sessionState.newHadoopConf() + // scalastyle:on deltahadoopconfiguration + spark.sparkContext.broadcast(new SerializableConfiguration(conf)) + } + + test("listing _change_data as a separate branch yields the same files and directories") { + withTempDir { tempDir => + // A table-shaped tree: data partitions plus a _change_data directory that itself mirrors the + // partitioning and nests deeper. Listing _change_data as its own branch (and excluding it + // from the main tree) must produce exactly the same set as listing everything inline. + val base = tempDir.getAbsolutePath + def mkFile(rel: String): Unit = { + val f = new File(base, rel) + f.getParentFile.mkdirs() + FileUtils.write(f, "x") + } + mkFile("f0.txt") + mkFile("part=1/f1.txt") + mkFile("part=2/f2.txt") + mkFile("_change_data/cdc0.txt") + mkFile("_change_data/part=1/cdc1.txt") + mkFile("_change_data/part=2/deep/cdc2.txt") + + val hadoopConf = broadcastHadoopConf() + def collectSet( + ds: org.apache.spark.sql.Dataset[SerializableFileStatus]): Set[(String, Boolean)] = + ds.collect().map(f => (f.path, f.isDir)).toSet + + // Everything listed inline (nothing hidden) -- the current behavior. + val inline = collectSet(DeltaFileOperations.recursiveListDirs( + spark, Seq(new Path(base).toString), hadoopConf, + hiddenDirNameFilter = _ => false, hiddenFileNameFilter = _ => false)) + + // Main tree with _change_data excluded, plus a separate _change_data branch (and its own + // directory entry) -- mirrors VacuumCommand.getFilesFromFilesystem. + implicit val statusEncoder: org.apache.spark.sql.Encoder[SerializableFileStatus] = + org.apache.spark.sql.Encoders.product[SerializableFileStatus] + val cdcPath = new Path(base, "_change_data") + val fs = cdcPath.getFileSystem(hadoopConf.value.value) + val mainOnly = DeltaFileOperations.recursiveListDirs( + spark, Seq(new Path(base).toString), hadoopConf, + hiddenDirNameFilter = (name: String) => name == "_change_data", + hiddenFileNameFilter = _ => false) + val cdcDir = spark.createDataset( + Seq(SerializableFileStatus.fromStatus(fs.getFileStatus(cdcPath)))) + val cdcListing = DeltaFileOperations.recursiveListDirs( + spark, Seq(cdcPath.toString), hadoopConf, + hiddenDirNameFilter = _ => false, hiddenFileNameFilter = _ => false) + val separate = collectSet(mainOnly.union(cdcDir).union(cdcListing)) + + assert(separate === inline, + s"only-in-separate=${separate -- inline}\nonly-in-inline=${inline -- separate}") + // Sanity check: the deeply nested _change_data file was actually discovered. + assert(inline.exists(_._1.endsWith("cdc2.txt")), s"deep CDC file missing: $inline") + } + } + + testFullVacuumOnly( + "VACUUM removes the same untracked _change_data files whether listed separately or inline") { + Seq(true, false).foreach { separately => + withSQLConf( + DeltaSQLConf.DELTA_VACUUM_LIST_CHANGE_DATA_DIR_SEPARATELY.key -> separately.toString) { + withEnvironment { (tempDir, _) => + val table = DeltaTableV2(spark, tempDir) + val committed = "committed.txt" + // Untracked change-data files, including one nested to mimic a partitioned CDF layout, so + // the separate _change_data branch must recurse past its first level to find it. + val cdcShallow = "_change_data/cdc-shallow.txt" + val cdcNested = "_change_data/part=1/deep/cdc-nested.txt" + val untrackedData = "data/untracked.txt" + gcTest(table, new ManualClock())( + CreateFile(committed, commitToActionLog = true), + CreateFile(cdcShallow, commitToActionLog = false), + CreateFile(cdcNested, commitToActionLog = false), + CreateFile(untrackedData, commitToActionLog = false), + CheckFiles(Seq(committed, cdcShallow, cdcNested, untrackedData)), + // The SQL VACUUM path uses the wall clock, so the epoch-0 files are past retention. + ExecuteVacuumInSQL(s"'$tempDir'", Seq(tempDir.toString)), + CheckFiles(Seq(committed)), + CheckFiles(Seq(cdcShallow, cdcNested, untrackedData), exist = false)) + } + } + } + } + testQuietly("basic case - SQL command on path-based tables with direct 'path'") { withEnvironment { (tempDir, _) => val table = DeltaTableV2(spark, tempDir)