From b1d14e7d061238631dd47f43e51719f19ebf015a Mon Sep 17 00:00:00 2001 From: Eunjin Song Date: Tue, 4 Aug 2026 12:09:04 -0700 Subject: [PATCH] [Spark] Add vacuum.listing.initialDepth to improve VACUUM listing parallelism VACUUM lists the table's file system by shallow-listing the table root, re-distributing that first level across the cluster, and then recursing each first-level directory in a single task. When the first level is skewed -- a low-cardinality partition column, or a large `_change_data` directory -- one task ends up listing an entire subtree while the rest of the cluster sits idle, which dominates VACUUM time on large tables. This adds `spark.databricks.delta.vacuum.listing.initialDepth` (default 1, which preserves the existing behavior exactly). Values greater than 1 make `recursiveListDirs` descend that many directory levels breadth-first, re-distributing the frontier after each level, before handing the remaining directories to the parallel subtree recursion. This yields a larger, better-spread set of directories to parallelize over. The set of files considered by VACUUM is unchanged for any depth; only the listing parallelism differs. A new test asserts this depth-invariance over a tree with files at multiple depths and nested empty directories, and an end-to-end VACUUM test confirms untracked files nested deeper than the configured depth are still discovered and removed. Addresses delta-io/delta#2201. Co-Authored-By: Claude Opus 4.8 --- .../sql/delta/commands/VacuumCommand.scala | 6 +- .../sql/delta/sources/DeltaSQLConf.scala | 14 +++ .../sql/delta/util/DeltaFileOperations.scala | 63 +++++++++++-- .../spark/sql/delta/DeltaVacuumSuite.scala | 93 ++++++++++++++++++- 4 files changed, 165 insertions(+), 11 deletions(-) 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..b54483331a5 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 @@ -912,6 +912,9 @@ trait VacuumCommandImpl extends DeltaCommand { ((_: String) => false, (_: String) => false) } + val initialListingDepth = + spark.sessionState.conf.getConf(DeltaSQLConf.DELTA_VACUUM_LISTING_INITIAL_DEPTH) + // Use DeltaFileOperations.recursiveListDirs val files = DeltaFileOperations.recursiveListDirs( spark, @@ -919,7 +922,8 @@ trait VacuumCommandImpl extends DeltaCommand { hadoopConf, hiddenDirNameFilter = hiddenDirFilter, hiddenFileNameFilter = hiddenFileFilter, - fileListingParallelism = parallelism + fileListingParallelism = parallelism, + initialListingDepth = initialListingDepth ) .map { f => // Make paths url-encoded (same pattern as VacuumCommand) 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..4f32268276b 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_LISTING_INITIAL_DEPTH = + buildConf("vacuum.listing.initialDepth") + .doc("The number of directory levels VACUUM lists shallowly (re-distributing the frontier " + + "across the cluster after each level) before fanning out to the parallel recursive " + + "listing. The default of 1 lists only the table's immediate children before fanning out, " + + "which can leave a single task listing an entire large subtree when the first level is " + + "skewed (e.g. a low-cardinality partition column, or the _change_data directory). " + + "Increasing this descends that many levels first so more directories are available to " + + "distribute, improving listing parallelism for such layouts. The set of files considered " + + "by VACUUM is unchanged; only the listing parallelism differs.") + .intConf + .checkValue(_ >= 1, "vacuum.listing.initialDepth must be at least 1") + .createWithDefault(1) + val ENFORCE_DELETED_FILE_AND_LOG_RETENTION_DURATION_COMPATIBILITY = buildConf("vacuum.enforceDeletedFileAndLogRetentionDurationCompatibility") .internal() diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala b/spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala index 6772f2bcaff..f2fb1b7d7cc 100644 --- a/spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala +++ b/spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala @@ -39,7 +39,9 @@ import org.apache.parquet.hadoop.{Footer, ParquetFileReader} import org.apache.spark.{SparkEnv, SparkException, TaskContext} import org.apache.spark.broadcast.Broadcast import org.apache.spark.internal.MDC +import org.apache.spark.rdd.RDD import org.apache.spark.sql.{Dataset, SparkSession} +import org.apache.spark.storage.StorageLevel import org.apache.spark.util.{SerializableConfiguration, ThreadUtils} /** @@ -230,6 +232,17 @@ object DeltaFileOperations extends DeltaLogging { * that are children to the path will be listed. If false, the paths are * treated as filenames, and files under the same folder with filenames * after the path will be listed instead. + * @param initialListingDepth The number of directory levels to list shallowly (one level per + * distributed round, re-distributing the frontier after each round) + * before handing the remaining directories to the parallel subtree + * recursion. Must be >= 1. The default of 1 preserves the original + * behavior: only the immediate children of `subDirs` are listed before + * fanning out, which can leave a single task listing an entire large + * subtree when the first level is skewed (e.g. a low-cardinality + * partition column, or `_change_data`). A larger value descends that + * many levels first so more directories are available to distribute. + * The listed set is identical regardless of this value; only the + * listing parallelism changes. */ def recursiveListDirs( spark: SparkSession, @@ -238,14 +251,19 @@ object DeltaFileOperations extends DeltaLogging { hiddenDirNameFilter: String => Boolean = defaultHiddenFileFilter, hiddenFileNameFilter: String => Boolean = defaultHiddenFileFilter, fileListingParallelism: Option[Int] = None, - listAsDirectories: Boolean = true): Dataset[SerializableFileStatus] = { + listAsDirectories: Boolean = true, + initialListingDepth: Int = 1): Dataset[SerializableFileStatus] = { import org.apache.spark.sql.delta.implicits._ if (subDirs.isEmpty) return spark.emptyDataset[SerializableFileStatus] - val listParallelism = fileListingParallelism.getOrElse(spark.sparkContext.defaultParallelism) - val subDirsParallelism = subDirs.length.min(spark.sparkContext.defaultParallelism) - val dirsAndFiles = spark.sparkContext.parallelize( - subDirs, - subDirsParallelism).mapPartitions { dirs => + require(initialListingDepth >= 1, + s"initialListingDepth must be >= 1, but got $initialListingDepth") + val sc = spark.sparkContext + val listParallelism = fileListingParallelism.getOrElse(sc.defaultParallelism) + val subDirsParallelism = subDirs.length.min(sc.defaultParallelism) + + // Level 0: shallow-list the supplied roots. `listAsDirectories` only applies to this level; + // every level below consists of real directories. + val firstLevel = sc.parallelize(subDirs, subDirsParallelism).mapPartitions { dirs => val logStore = LogStore(SparkEnv.get.conf, hadoopConf.value.value) listUsingLogStore( logStore, @@ -255,16 +273,43 @@ object DeltaFileOperations extends DeltaLogging { hiddenDirNameFilter, hiddenFileNameFilter, listAsDirectories) }.repartition(listParallelism) // Initial list of subDirs may be small - val allDirsAndFiles = dirsAndFiles.mapPartitions { firstLevelDirsAndFiles => + // Optionally descend `initialListingDepth - 1` further levels breadth-first, re-distributing + // the frontier after each level, so the parallel recursion below starts from a well-spread set + // of directories rather than a possibly skewed first level. Files, and interior directories + // whose children are listed here, are emitted as they are found; only the final frontier of + // directories is passed to the subtree recursion (which emits those directories itself). + var frontier = firstLevel + var collected: RDD[SerializableFileStatus] = sc.emptyRDD[SerializableFileStatus] + var levelsToDescend = initialListingDepth - 1 + while (levelsToDescend > 0) { + // Consumed twice below (emitted + re-listed), so avoid re-hitting the file system. + frontier.persist(StorageLevel.MEMORY_AND_DISK) + val interiorDirs = frontier.filter(_.isDir) + collected = collected + .union(frontier.filter(f => !f.isDir)) // terminal files at this level + .union(interiorDirs) // interior directories, emitted exactly once + frontier = interiorDirs.map(_.path).mapPartitions { dirs => + val logStore = LogStore(SparkEnv.get.conf, hadoopConf.value.value) + listUsingLogStore( + logStore, + hadoopConf.value.value, + dirs, + recurse = false, + hiddenDirNameFilter, hiddenFileNameFilter) + }.repartition(listParallelism) + levelsToDescend -= 1 + } + + val allDirsAndFiles = frontier.mapPartitions { frontierDirsAndFiles => val logStore = LogStore(SparkEnv.get.conf, hadoopConf.value.value) recurseDirectories( logStore, hadoopConf.value.value, - firstLevelDirsAndFiles, + frontierDirsAndFiles, hiddenDirNameFilter, hiddenFileNameFilter) } - spark.createDataset(allDirsAndFiles) + spark.createDataset(collected.union(allDirsAndFiles)) } /** 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..552aa67ef40 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,97 @@ 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("recursiveListDirs returns the same files and directories for any initialListingDepth") { + withTempDir { tempDir => + // A tree with files at multiple depths and empty directories, including a nested only-child + // empty chain (d/e) and an empty directory beside files (a/emptyB). The set of listed + // files/directories must not depend on how many levels are listed before fanning out. + val base = tempDir.getAbsolutePath + def mkFile(rel: String): Unit = { + val f = new File(base, rel) + f.getParentFile.mkdirs() + FileUtils.write(f, "x") + } + def mkDir(rel: String): Unit = assert(new File(base, rel).mkdirs()) + + mkFile("f0.txt") + mkFile("a/f1.txt") + mkFile("a/b/f2.txt") + mkFile("a/b/c/f3.txt") + mkDir("a/emptyB") + mkDir("d/e") + mkDir("g") + + val hadoopConf = broadcastHadoopConf() + def listAt(depth: Int): Set[(String, Boolean)] = DeltaFileOperations.recursiveListDirs( + spark, + Seq(new Path(base).toString), + hadoopConf, + hiddenDirNameFilter = _ => false, + hiddenFileNameFilter = _ => false, + initialListingDepth = depth) + .collect() + .map(f => (f.path, f.isDir)) + .toSet + + val expected = listAt(1) + // Sanity-check that the whole tree (4 files, 7 directories) was actually discovered. + assert(expected.count(!_._2) === 4, s"expected 4 files, got $expected") + assert(expected.count(_._2) === 7, s"expected 7 directories, got $expected") + + // Identical result for every depth, including depths deeper than the tree itself. + (2 to 5).foreach { depth => + assert(listAt(depth) === expected, + s"listing at initialListingDepth=$depth differed from initialListingDepth=1") + } + } + } + + test("recursiveListDirs rejects a non-positive initialListingDepth") { + withTempDir { tempDir => + val e = intercept[IllegalArgumentException] { + DeltaFileOperations.recursiveListDirs( + spark, + Seq(new Path(tempDir.getAbsolutePath).toString), + broadcastHadoopConf(), + initialListingDepth = 0) + } + assert(e.getMessage.contains("initialListingDepth must be >= 1")) + } + } + + testFullVacuumOnly( + "VACUUM with a larger vacuum.listing.initialDepth deletes the same untracked files") { + withSQLConf(DeltaSQLConf.DELTA_VACUUM_LISTING_INITIAL_DEPTH.key -> "3") { + withEnvironment { (tempDir, _) => + val table = DeltaTableV2(spark, tempDir) + val committed = "committed.txt" + val untrackedShallow = "sub/untrackedShallow.txt" + // Nested deeper than initialDepth=3 so the recursive fan-out past the shallow-listed + // levels must still reach it. + val untrackedDeep = "sub/w/x/y/untrackedDeep.txt" + gcTest(table, new ManualClock())( + CreateFile(committed, commitToActionLog = true), + CreateFile(untrackedShallow, commitToActionLog = false), + CreateFile(untrackedDeep, commitToActionLog = false), + CheckFiles(Seq(committed, untrackedShallow, untrackedDeep)), + // The SQL VACUUM path uses the wall clock, so the epoch-0 files are past retention. A + // depth of 3 must still discover the deeply nested untracked file. + ExecuteVacuumInSQL(s"'$tempDir'", Seq(tempDir.toString)), + CheckFiles(Seq(committed)), + CheckFiles(Seq(untrackedShallow, untrackedDeep), exist = false)) + } + } + } + testQuietly("basic case - SQL command on path-based tables with direct 'path'") { withEnvironment { (tempDir, _) => val table = DeltaTableV2(spark, tempDir)