This is an automated email from the ASF dual-hosted git repository.
LuciferYang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 2f1536def224 [SPARK-57530][CORE] Make SparkFileUtils.recursiveList
null-safe and linear-time
2f1536def224 is described below
commit 2f1536def224155eb48d5e5263735dbe7644b2d9
Author: YangJie <[email protected]>
AuthorDate: Tue Jun 30 17:10:30 2026 +0800
[SPARK-57530][CORE] Make SparkFileUtils.recursiveList null-safe and
linear-time
### What changes were proposed in this pull request?
`SparkFileUtils.recursiveList` walks a directory tree by reading
`File.listFiles` and draining a buffer used as a work queue. Two problems:
- It calls `listFiles` without a null check. `File.listFiles` returns null
when a directory cannot be read (an IO error, or the directory being removed
during the walk), so the walk could throw `NullPointerException`.
- It takes the next directory to visit with `Buffer.remove(0)`, which is
O(n); on a wide tree the whole walk is O(n^2).
This PR rewrites the traversal to:
- null-guard both `listFiles` calls, skipping (and logging) a directory
that cannot be listed instead of throwing;
- use a `Queue` with O(1) dequeue, so the walk is linear in the number of
entries.
It also drops the now-clearly-dead `Option(...).getOrElse(Array.empty)`
guard at the `RocksDBFileManager` call site, since `recursiveList` never
returns null. For a readable tree the set of returned entries is unchanged.
### Why are the changes needed?
`recursiveList` runs on real paths, such as the RocksDB state-store file
manager and `LocalSparkCluster`. A directory that becomes unreadable mid-walk
would crash the walk with an NPE instead of returning what it could, and the
O(n^2) `remove(0)` is needless overhead on directories with many entries.
### Does this PR introduce _any_ user-facing change?
No. For a readable directory tree the returned entries are identical; the
only difference is that a directory which cannot be listed is now skipped with
a warning instead of raising an internal `NullPointerException`.
### How was this patch tested?
Added `SparkFileUtilsSuite` with three cases: a nested listing, a root
whose `listFiles` returns null, and a subdirectory whose `listFiles` returns
null mid-walk (with a spy confirming non-directories are not recursed into).
The two null cases throw on the old code and pass with the fix. `build/sbt
'common-utils/testOnly *SparkFileUtilsSuite'` passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56590 from LuciferYang/SPARK-recursivelist-robustness.
Authored-by: YangJie <[email protected]>
Signed-off-by: yangjie01 <[email protected]>
---
.../org/apache/spark/util/SparkFileUtils.scala | 29 ++++++--
common/utils/src/test/resources/log4j2.properties | 12 +++
.../apache/spark/util/SparkFileUtilsSuite.scala | 86 ++++++++++++++++++++++
.../streaming/state/RocksDBFileManager.scala | 2 +-
4 files changed, 121 insertions(+), 8 deletions(-)
diff --git
a/common/utils/src/main/scala/org/apache/spark/util/SparkFileUtils.scala
b/common/utils/src/main/scala/org/apache/spark/util/SparkFileUtils.scala
index 3f1f9c1f9df7..7124ca30bc61 100644
--- a/common/utils/src/main/scala/org/apache/spark/util/SparkFileUtils.scala
+++ b/common/utils/src/main/scala/org/apache/spark/util/SparkFileUtils.scala
@@ -21,6 +21,8 @@ import java.net.{URI, URISyntaxException, URL}
import java.nio.file.{Files, Path, StandardCopyOption}
import java.nio.file.attribute.FileTime
+import scala.collection.mutable
+
import org.apache.spark.internal.{Logging, LogKeys}
import org.apache.spark.network.util.JavaUtils
@@ -59,16 +61,29 @@ private[spark] trait SparkFileUtils extends Logging {
/**
* Lists files recursively.
+ *
+ * A directory that cannot be listed (`File.listFiles` returns null, e.g. an
IO error or the
+ * directory being removed during the walk) is skipped with a warning rather
than throwing, so
+ * the result may be partial when part of the tree is unreadable.
*/
def recursiveList(f: File): Array[File] = {
require(f.isDirectory)
- val result = f.listFiles.toBuffer
- val dirList = result.filter(_.isDirectory)
- while (dirList.nonEmpty) {
- val curDir = dirList.remove(0)
- val files = curDir.listFiles()
- result ++= files
- dirList ++= files.filter(_.isDirectory)
+ val result = mutable.ArrayBuffer[File]()
+ // Use a queue with O(1) dequeue rather than removing from the head of a
buffer (O(n)), so the
+ // walk stays linear in the number of entries.
+ val dirs = mutable.Queue[File](f)
+ while (dirs.nonEmpty) {
+ val dir = dirs.dequeue()
+ // `File.listFiles` returns null when the directory cannot be read (an
IO error, or it was
+ // removed during the walk); skip it instead of throwing an NPE, but log
it so a partial
+ // result is traceable.
+ val entries = dir.listFiles()
+ if (entries != null) {
+ result ++= entries
+ dirs ++= entries.filter(_.isDirectory)
+ } else {
+ logWarning(log"Failed to list directory ${MDC(LogKeys.PATH, dir)};
skipping it.")
+ }
}
result.toArray
}
diff --git a/common/utils/src/test/resources/log4j2.properties
b/common/utils/src/test/resources/log4j2.properties
index cb38f5b55a0b..e23c60e58ae0 100644
--- a/common/utils/src/test/resources/log4j2.properties
+++ b/common/utils/src/test/resources/log4j2.properties
@@ -38,6 +38,13 @@ appender.pattern.fileName = target/pattern.log
appender.pattern.layout.type = PatternLayout
appender.pattern.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n%ex
+# SparkFileUtils Logging Appender (used by SparkFileUtilsSuite to assert on
warnings)
+appender.spark_file_utils.type = File
+appender.spark_file_utils.name = spark_file_utils
+appender.spark_file_utils.fileName = target/spark-file-utils.log
+appender.spark_file_utils.layout.type = PatternLayout
+appender.spark_file_utils.layout.pattern = %d{yy/MM/dd HH:mm:ss} %p %c{1}:
%m%n%ex
+
# Custom loggers
logger.structured_logging.name = org.apache.spark.util.StructuredLoggingSuite
logger.structured_logging.level = trace
@@ -58,3 +65,8 @@ logger.pattern_logger.name =
org.apache.spark.util.PatternSparkLoggerSuite
logger.pattern_logger.level = trace
logger.pattern_logger.appenderRefs = pattern
logger.pattern_logger.appenderRef.pattern.ref = pattern
+
+logger.spark_file_utils.name = org.apache.spark.util.SparkFileUtils
+logger.spark_file_utils.level = trace
+logger.spark_file_utils.appenderRefs = spark_file_utils
+logger.spark_file_utils.appenderRef.spark_file_utils.ref = spark_file_utils
diff --git
a/common/utils/src/test/scala/org/apache/spark/util/SparkFileUtilsSuite.scala
b/common/utils/src/test/scala/org/apache/spark/util/SparkFileUtilsSuite.scala
new file mode 100644
index 000000000000..02b09dbc24d2
--- /dev/null
+++
b/common/utils/src/test/scala/org/apache/spark/util/SparkFileUtilsSuite.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.util
+
+import java.io.File
+import java.nio.file.Files
+
+import org.scalatest.funsuite.AnyFunSuite // scalastyle:ignore funsuite
+
+class SparkFileUtilsSuite extends AnyFunSuite { // scalastyle:ignore funsuite
+
+ // Returns the log content appended to `target/spark-file-utils.log`
(configured for the
+ // SparkFileUtils logger in log4j2.properties) while running `f`.
+ private def captureLogOutput(f: () => Unit): String = {
+ val logFile = new File(new File(".").getCanonicalPath,
"target/spark-file-utils.log")
+ val before = if (logFile.exists()) Files.readString(logFile.toPath) else ""
+ f()
+ val after = if (logFile.exists()) Files.readString(logFile.toPath) else ""
+ after.substring(before.length)
+ }
+
+ test("recursiveList lists all nested files and directories") {
+ val root = Files.createTempDirectory("spark-recursive-list").toFile
+ try {
+ val sub = new File(root, "sub")
+ assert(sub.mkdir())
+ val nested = new File(sub, "nested")
+ assert(nested.mkdir())
+ val topFile = new File(root, "top.txt")
+ assert(topFile.createNewFile())
+ val subFile = new File(sub, "sub.txt")
+ assert(subFile.createNewFile())
+
+ assert(SparkFileUtils.recursiveList(root).toSet === Set(sub, nested,
topFile, subFile))
+ } finally {
+ SparkFileUtils.deleteQuietly(root)
+ }
+ }
+
+ test("recursiveList returns empty and warns instead of throwing when a dir
cannot be listed") {
+ // A directory whose listFiles returns null must yield an empty result,
not an NPE, and the
+ // skipped directory must be logged so a partial result is traceable.
+ val unreadable = new File("spark-unreadable-dir") {
+ override def isDirectory: Boolean = true
+ override def listFiles(): Array[File] = null
+ }
+ val logOutput = captureLogOutput(() =>
assert(SparkFileUtils.recursiveList(unreadable).isEmpty))
+ assert(logOutput.contains("Failed to list directory"))
+ assert(logOutput.contains("spark-unreadable-dir"))
+ }
+
+ test("recursiveList skips a subdirectory whose listFiles returns null
mid-walk") {
+ // The null directory is not the root but one discovered during the walk,
so the guard must
+ // hold at depth > 0. `unreadableSub` is a directory whose listFiles
returns null; `leaf` is a
+ // plain file that must be returned but never recursed into (a spy
verifies that).
+ var leafListed = false
+ val leaf = new File("leaf.txt") {
+ override def isDirectory: Boolean = false
+ override def listFiles(): Array[File] = { leafListed = true;
super.listFiles() }
+ }
+ val unreadableSub = new File("unreadable-sub") {
+ override def isDirectory: Boolean = true
+ override def listFiles(): Array[File] = null
+ }
+ val root = new File("root") {
+ override def isDirectory: Boolean = true
+ override def listFiles(): Array[File] = Array(leaf, unreadableSub)
+ }
+ assert(SparkFileUtils.recursiveList(root).toSet === Set(leaf,
unreadableSub))
+ assert(!leafListed, "a non-directory entry must not be recursed into")
+ }
+}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
index 374cfc5ed0cd..8284f50ec1a8 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDBFileManager.scala
@@ -1086,7 +1086,7 @@ class RocksDBFileManager(
/** Log the files present in a directory. This is useful for debugging. */
private def logFilesInDir(dir: File, msg: MessageWithContext): Unit = {
- lazy val files =
Option(Utils.recursiveList(dir)).getOrElse(Array.empty).map { f =>
+ lazy val files = Utils.recursiveList(dir).map { f =>
s"${f.getAbsolutePath} - ${f.length()} bytes"
}
logDebug(msg + log" - ${MDC(LogKeys.NUM_FILES, files.length)} files\n\t" +
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
