[Spark] Add vacuum.listing.initialDepth to improve VACUUM listing parallelism

spark/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala

@@ -912,14 +912,18 @@ 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,
       Seq(basePath),
       hadoopConf,
       hiddenDirNameFilter = hiddenDirFilter,
       hiddenFileNameFilter = hiddenFileFilter,
-      fileListingParallelism = parallelism
+      fileListingParallelism = parallelism,
+      initialListingDepth = initialListingDepth
     )
       .map { f =>
         // Make paths url-encoded (same pattern as VacuumCommand)

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()

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))
   }
 
   /**

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)