[SPARK-53073][CORE][SQL][YARN][SS] Support copyDirectory in SparkFileUtils and JavaUtils

### What changes were proposed in this pull request?

This PR aims to support `copyDirectory` in `SparkFileUtils` and `JavaUtils`.

### Why are the changes needed?

To provide a better implementation.

**BEFORE**

```scala
scala> spark.time(org.apache.commons.io.FileUtils.copyDirectory(new java.io.File("/tmp/spark"), new java.io.File("/tmp/spark2")))
Time taken: 5128 ms
```

**AFTER**

```scala
scala> spark.time(org.apache.spark.network.util.JavaUtils.copyDirectory(new java.io.File("/tmp/spark"), new java.io.File("/tmp/spark2")))
Time taken: 2979 ms
```

### Does this PR introduce _any_ user-facing change?

No behavior change.

### How was this patch tested?

Pass the CIs.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#51786 from dongjoon-hyun/SPARK-53073.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: dongjoon-hyun

common/utils/src/main/java/org/apache/spark/network/util/JavaUtils.java

@@ -26,6 +26,7 @@
 import java.nio.file.Path;
 import java.nio.file.FileVisitResult;
 import java.nio.file.SimpleFileVisitor;
+import java.nio.file.StandardCopyOption;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
@@ -74,6 +75,34 @@ public static void deleteQuietly(File file) {
     }
   }
 
+  /** Copy src to the target directory simply. File attribute times are not copied. */
+  public static void copyDirectory(File src, File dst) throws IOException {
+    if (src == null || dst == null || !src.exists() || !src.isDirectory() ||
+        (dst.exists() && !dst.isDirectory())) {
+      throw new IllegalArgumentException("Invalid input file " + src + " or directory " + dst);
+    }
+    Path from = src.toPath().toAbsolutePath().normalize();
+    Path to = dst.toPath().toAbsolutePath().normalize();
+    if (to.startsWith(from)) {
+       throw new IllegalArgumentException("Cannot copy directory to itself or its subdirectory");
+    }
+    Files.createDirectories(to);
+    Files.walkFileTree(from, new SimpleFileVisitor<Path>() {
+      @Override
+      public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs)
+          throws IOException {
+        Files.createDirectories(to.resolve(from.relativize(dir)));
+        return FileVisitResult.CONTINUE;
+      }
+
+      @Override
+      public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
+        Files.copy(file, to.resolve(from.relativize(file)), StandardCopyOption.REPLACE_EXISTING);
+        return FileVisitResult.CONTINUE;
+      }
+    });
+  }
+
   /** Returns a hash consistent with Spark's Utils.nonNegativeHash(). */
   public static int nonNegativeHash(Object obj) {
     if (obj == null) { return 0; }

common/utils/src/main/scala/org/apache/spark/util/SparkFileUtils.scala

@@ -147,6 +147,11 @@ private[spark] trait SparkFileUtils extends Logging {
     }.toFile
   }
 
+  /** Copy src to the target directory simply. File attribute times are not copied. */
+  def copyDirectory(src: File, dir: File): Unit = {
+    JavaUtils.copyDirectory(src, dir)
+  }
+
   /** Copy file to the target directory simply. File attribute times are not copied. */
   def copyFileToDirectory(file: File, dir: File): Unit = {
     if (file == null || dir == null || !file.exists() || (dir.exists() && !dir.isDirectory())) {

connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala

@@ -29,7 +29,6 @@ import scala.io.Source
 import scala.jdk.CollectionConverters._
 import scala.util.Random
 
-import org.apache.commons.io.FileUtils
 import org.apache.kafka.clients.producer.{ProducerRecord, RecordMetadata}
 import org.apache.kafka.common.TopicPartition
 import org.scalatest.concurrent.PatienceConfiguration.Timeout
@@ -1729,7 +1728,7 @@ abstract class KafkaMicroBatchV2SourceSuite extends KafkaMicroBatchSourceSuiteBa
     val checkpointDir = Utils.createTempDir().getCanonicalFile
     // Copy the checkpoint to a temp dir to prevent changes to the original.
     // Not doing this will lead to the test passing on the first run, but fail subsequent runs.
-    FileUtils.copyDirectory(new File(resourceUri), checkpointDir)
+    Utils.copyDirectory(new File(resourceUri), checkpointDir)
 
     testStream(query)(
       StartStream(checkpointLocation = checkpointDir.getAbsolutePath),

resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnShuffleIntegrationSuite.scala

@@ -21,7 +21,6 @@ import java.io.File
 import java.nio.charset.StandardCharsets
 
 import com.google.common.io.Files
-import org.apache.commons.io.FileUtils
 import org.apache.hadoop.yarn.conf.YarnConfiguration
 import org.scalatest.matchers.must.Matchers
 import org.scalatest.matchers.should.Matchers._
@@ -173,7 +172,7 @@ private object YarnExternalShuffleDriver extends Logging with Matchers {
         val dbBackend = DBBackend.byName(dbBackendName)
         logWarning(s"Use ${dbBackend.name()} as the implementation of " +
           s"${SHUFFLE_SERVICE_DB_BACKEND.key}")
-        FileUtils.copyDirectory(registeredExecFile, execStateCopy)
+        Utils.copyDirectory(registeredExecFile, execStateCopy)
         assert(!ShuffleTestAccessor
           .reloadRegisteredExecutors(dbBackend, execStateCopy).isEmpty)
       }

scalastyle-config.xml

@@ -322,6 +322,11 @@ This file is divided into 3 sections:
     <customMessage>Use copyFileToDirectory of SparkFileUtils or Utils instead.</customMessage>
   </check>
 
+  <check customId="copyDirectory" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
+    <parameters><parameter name="regex"> FileUtils\.copyDirectory</parameter></parameters>
+    <customMessage>Use copyDirectory of JavaUtils/SparkFileUtils/Utils instead.</customMessage>
+  </check>
+
   <check customId="commonslang2" level="error" class="org.scalastyle.file.RegexChecker" enabled="true">
     <parameters><parameter name="regex">org\.apache\.commons\.lang\.</parameter></parameters>
     <customMessage>Use Commons Lang 3 classes (package org.apache.commons.lang3.*) instead

sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/client/ClassFinderSuite.scala

@@ -18,8 +18,6 @@ package org.apache.spark.sql.connect.client
 
 import java.nio.file.Paths
 
-import org.apache.commons.io.FileUtils
-
 import org.apache.spark.sql.connect.test.ConnectFunSuite
 import org.apache.spark.util.SparkFileUtils
 
@@ -32,7 +30,7 @@ class ClassFinderSuite extends ConnectFunSuite {
     requiredClasses.foreach(className =>
       assume(classResourcePath.resolve(className).toFile.exists))
     val copyDir = SparkFileUtils.createTempDir().toPath
-    FileUtils.copyDirectory(classResourcePath.toFile, copyDir.toFile)
+    SparkFileUtils.copyDirectory(classResourcePath.toFile, copyDir.toFile)
     val monitor = new REPLClassDirMonitor(copyDir.toAbsolutePath.toString)
 
     def checkClasses(monitor: REPLClassDirMonitor, additionalClasses: Seq[String] = Nil): Unit = {

sql/core/src/main/scala/org/apache/spark/sql/artifact/ArtifactManager.scala

@@ -340,7 +340,7 @@ class ArtifactManager(session: SparkSession) extends AutoCloseable with Logging
     val sparkContext = session.sparkContext
     val newArtifactManager = new ArtifactManager(newSession)
     if (artifactPath.toFile.exists()) {
-      FileUtils.copyDirectory(artifactPath.toFile, newArtifactManager.artifactPath.toFile)
+      Utils.copyDirectory(artifactPath.toFile, newArtifactManager.artifactPath.toFile)
     }
     val blockManager = sparkContext.env.blockManager
     val newBlockIds = cachedBlockIdList.asScala.map { blockId =>

sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala

@@ -25,8 +25,6 @@ import java.util.concurrent.atomic.AtomicBoolean
 import scala.collection.mutable.HashSet
 import scala.concurrent.duration._
 
-import org.apache.commons.io.FileUtils
-
 import org.apache.spark.{CleanerListener, SparkRuntimeException}
 import org.apache.spark.executor.DataReadMethod._
 import org.apache.spark.executor.DataReadMethod.DataReadMethod
@@ -1405,7 +1403,7 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
         .filter(_.startsWith("Location:"))
         .head
         .replace("Location: file:", "")
-      FileUtils.copyDirectory(
+      Utils.copyDirectory(
         new File(part0Loc),
         new File(part0Loc.replace("part=0", "part=1")))
 

sql/core/src/test/scala/org/apache/spark/sql/artifact/ArtifactManagerSuite.scala

@@ -56,7 +56,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
     assume(artifactPath.resolve("smallClassFile.class").toFile.exists)
 
     val copyDir = Utils.createTempDir().toPath
-    FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+    Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
     val stagingPath = copyDir.resolve("smallClassFile.class")
     assert(stagingPath.toFile.exists())
     val remotePath = Paths.get("classes/smallClassFile.class")
@@ -72,7 +72,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
     assume(artifactPath.resolve("Hello.class").toFile.exists)
 
     val copyDir = Utils.createTempDir().toPath
-    FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+    Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
     val stagingPath = copyDir.resolve("Hello.class")
     assert(stagingPath.toFile.exists())
     val remotePath = Paths.get("classes/Hello.class")
@@ -98,7 +98,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
     assume(artifactPath.resolve("Hello.class").toFile.exists)
 
     val copyDir = Utils.createTempDir().toPath
-    FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+    Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
     val stagingPath = copyDir.resolve("Hello.class")
     assert(stagingPath.toFile.exists())
     val remotePath = Paths.get("classes/Hello.class")
@@ -180,7 +180,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
 
     val copyDir = Utils.createTempDir().toPath
     val destFSDir = Utils.createTempDir().toPath
-    FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+    Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
     val stagingPath = copyDir.resolve("smallClassFile.class")
     val remotePath = Paths.get("forward_to_fs", destFSDir.toString, "smallClassFileCopied.class")
     assert(stagingPath.toFile.exists())
@@ -203,7 +203,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
       val blockId = CacheId(spark.sessionUUID, "abc")
       // Setup artifact dir
       val copyDir = Utils.createTempDir().toPath
-      FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+      Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
       try {
         artifactManager.addArtifact(remotePath, stagingPath, None)
         val stagingPathFile = copyDir.resolve("smallClassFile.class")
@@ -246,7 +246,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
 
     def addHelloClass(session: SparkSession): Unit = {
       val copyDir = Utils.createTempDir().toPath
-      FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+      Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
       val stagingPath = copyDir.resolve("Hello.class")
       val remotePath = Paths.get("classes/Hello.class")
       assert(stagingPath.toFile.exists())
@@ -299,7 +299,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
     assume(artifactPath.resolve("Hello.class").toFile.exists)
 
     val copyDir = Utils.createTempDir().toPath
-    FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+    Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
     val stagingPath = copyDir.resolve("Hello.class")
     val remotePath = Paths.get("classes/Hello.class")
 
@@ -368,7 +368,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
     val copyDir = Utils.createTempDir().toPath
     assume(artifactPath.resolve(classFileToUse).toFile.exists)
 
-    FileUtils.copyDirectory(artifactPath.toFile, copyDir.toFile)
+    Utils.copyDirectory(artifactPath.toFile, copyDir.toFile)
     val classPath = copyDir.resolve(classFileToUse)
     assert(classPath.toFile.exists())
 
@@ -407,7 +407,7 @@ class ArtifactManagerSuite extends SharedSparkSession {
     withTempPath { dir =>
       val path = dir.toPath
       // Setup artifact dir
-      FileUtils.copyDirectory(artifactPath.toFile, dir)
+      Utils.copyDirectory(artifactPath.toFile, dir)
       val randomFilePath = path.resolve("random_file")
       val testBytes = "test".getBytes(StandardCharsets.UTF_8)
       Files.write(randomFilePath, testBytes)

sql/core/src/test/scala/org/apache/spark/sql/execution/command/DDLCommandTestUtils.scala

@@ -19,7 +19,6 @@ package org.apache.spark.sql.execution.command
 
 import java.io.File
 
-import org.apache.commons.io.FileUtils
 import org.apache.hadoop.fs.{FileSystem, Path}
 import org.scalactic.source.Position
 import org.scalatest.Tag
@@ -29,6 +28,7 @@ import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.connector.catalog.{CatalogV2Util, TableCatalog}
 import org.apache.spark.sql.execution.datasources.PartitioningUtils
 import org.apache.spark.sql.test.SQLTestUtils
+import org.apache.spark.util.Utils
 
 /**
  * The common settings and utility functions for all v1 and v2 test suites. When a function
@@ -170,7 +170,7 @@ trait DDLCommandTestUtils extends SQLTestUtils {
   def copyPartition(tableName: String, from: String, to: String): String = {
     val part0Loc = getPartitionLocation(tableName, from)
     val part1Loc = part0Loc.replace(from, to)
-    FileUtils.copyDirectory(new File(part0Loc), new File(part1Loc))
+    Utils.copyDirectory(new File(part0Loc), new File(part1Loc))
     part1Loc
   }