add listLeafDirStatuses and listLeafDirStatusesback to SparkHadoopUtil.scala

Author: anuvedverma

core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala

@@ -21,13 +21,16 @@ import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, Da
 import java.net.InetAddress
 import java.security.PrivilegedExceptionAction
 import java.text.DateFormat
-import java.util.{Date, Locale}
+import java.util.{Arrays, Date, Locale}
 
 import scala.collection.JavaConverters._
+import scala.collection.immutable.Map
 import scala.collection.mutable
 import scala.collection.mutable.HashMap
 import scala.language.existentials
+import scala.util.control.NonFatal
 
+import com.google.common.primitives.Longs
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs._
 import org.apache.hadoop.hdfs.DistributedFileSystem.HdfsDataOutputStreamBuilder
@@ -225,6 +228,27 @@ private[spark] class SparkHadoopUtil extends Logging {
     if (baseStatus.isDirectory) recurse(baseStatus) else Seq(baseStatus)
   }
 
+  /**
+   * [LYFT-INTERNAL] Removed from OSS Spark: https://github.com/apache/spark/pull/40942/
+   */
+  def listLeafDirStatuses(fs: FileSystem, basePath: Path): Seq[FileStatus] = {
+    listLeafDirStatuses(fs, fs.getFileStatus(basePath))
+  }
+
+  /**
+   * [LYFT-INTERNAL] Removed from OSS Spark: https://github.com/apache/spark/pull/40942/
+   */
+  def listLeafDirStatuses(fs: FileSystem, baseStatus: FileStatus): Seq[FileStatus] = {
+    def recurse(status: FileStatus): Seq[FileStatus] = {
+      val (directories, files) = fs.listStatus(status.getPath).partition(_.isDirectory)
+      val leaves = if (directories.isEmpty) Seq(status) else Seq.empty[FileStatus]
+      leaves ++ directories.flatMap(dir => listLeafDirStatuses(fs, dir))
+    }
+
+    assert(baseStatus.isDirectory)
+    recurse(baseStatus)
+  }
+
   def isGlobPath(pattern: Path): Boolean = {
     pattern.toString.exists("{}[]*?\\".toSet.contains)
   }