This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 0417f91069cd [SPARK-57354][SQL] Add ignoredPathSegmentRegex data
source option and config
0417f91069cd is described below
commit 0417f91069cd216783ea0a4c968b03125c2a2622
Author: Calvin Qin <[email protected]>
AuthorDate: Tue Jun 23 13:28:40 2026 -0700
[SPARK-57354][SQL] Add ignoredPathSegmentRegex data source option and config
### What changes were proposed in this pull request?
Add an `ignoredPathSegmentRegex` data source option plus a
`spark.sql.files.ignoredPathSegmentRegex` session config (default `^[._]`) for
file-based sources. The value is a Java regex matched with find semantics
against each individual directory/file name during listing; matching names are
skipped from file listing, partition discovery, and reads (a matching directory
excludes its whole subtree). The per-read option overrides the session config.
Both batch and Structured Streaming rea [...]
Regardless of the regex, three rules are hardcoded:
- `_metadata/_common_metadata` (Parquet summary files) are always listed
- `*._COPYING_` (in-flight Hadoop FS-shell copies) are always skipped
- `_-`prefixed partition dirs containing `=` are always kept.
A regex that never matches (`(?!)`, or empty string -- pending decision
below) disables the generic hidden-file filter.
Two carve-outs: table statistics (CommandUtils) pin the default regex so
ANALYZE sizes are unaffected by the session conf; and Parquet schema inference
now excludes zero-length files from footer candidates so a directory containing
a 0-byte _SUCCESS marker stays readable when hidden files are surfaced.
User documentation is added to `sql-data-sources-generic-options.md` and
the Structured Streaming guide.
### Why are the changes needed?
Hidden files listing is being requested as a feature. We want to enable a
user-facing option for this.
### Does this PR introduce _any_ user-facing change?
Yes. By setting the new option `ignorePathSegmentRegex = ""`, hidden files
will now show up inside all file listing operations. Default value preserves
current behavior of skipping hidden files.
### How was this patch tested?
Added unit tests covering the data source option and session conf,
exercising the file-listing path, partition discovery, batch reads, and
structured streaming. Also added tests for the session conf in streaming,
partition discovery with a hidden directory next to partition dirs, and reading
an unmodified Spark-written parquet directory with the option enabled.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 4.8
Closes #56374 from CalvQ/list-hidden-files-option.
Lead-authored-by: Calvin Qin <[email protected]>
Co-authored-by: Calvin Qin <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/util/HadoopFSUtils.scala | 100 +++++++-----
.../org/apache/spark/util/HadoopFSUtilsSuite.scala | 132 ++++++++++++++-
docs/sql-data-sources-generic-options.md | 40 +++++
docs/sql-migration-guide.md | 1 +
docs/streaming/apis-on-dataframes-and-datasets.md | 2 +
.../examples/sql/JavaSQLDataSourceExample.java | 7 +
examples/src/main/python/sql/datasource.py | 8 +
examples/src/main/r/RSparkSQLExample.R | 6 +
.../spark/examples/sql/SQLDataSourceExample.scala | 7 +
.../spark/sql/catalyst/FileSourceOptions.scala | 41 ++++-
.../org/apache/spark/sql/internal/SQLConf.scala | 26 ++-
.../spark/sql/execution/command/CommandUtils.scala | 11 +-
.../sql/execution/datasources/ArchiveReader.scala | 33 ++--
.../sql/execution/datasources/DataSource.scala | 14 +-
.../execution/datasources/FileIndexOptions.scala | 1 +
.../execution/datasources/InMemoryFileIndex.scala | 29 +++-
.../datasources/PartitioningAwareFileIndex.scala | 20 ++-
.../execution/datasources/csv/CSVDataSource.scala | 29 ++--
.../execution/datasources/csv/CSVFileFormat.scala | 5 +-
.../datasources/parquet/ParquetUtils.scala | 5 +-
.../sql/execution/datasources/v2/FileTable.scala | 6 +-
.../spark/sql/FileBasedDataSourceSuite.scala | 164 +++++++++++++++++++
.../execution/datasources/ArchiveReaderSuite.scala | 16 ++
.../sql/execution/datasources/FileIndexSuite.scala | 179 ++++++++++++++++++++-
.../sql/streaming/FileStreamSourceSuite.scala | 79 +++++++++
25 files changed, 880 insertions(+), 81 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
b/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
index f84184f7c1fa..2f5b02dbf577 100644
--- a/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/HadoopFSUtils.scala
@@ -18,9 +18,9 @@
package org.apache.spark.util
import java.io.FileNotFoundException
+import java.util.regex.Pattern
import scala.collection.mutable
-import scala.util.matching.Regex
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs._
@@ -37,6 +37,18 @@ import org.apache.spark.util.ArrayImplicits._
* Utility functions to simplify and speed-up file listing.
*/
private[spark] object HadoopFSUtils extends Logging {
+ /**
+ * The default value of the `ignoredPathSegmentRegex` option: a regex
evaluated with find
+ * semantics against each individual directory and file name during file
listing. It hides names
+ * that start with '_' or '.'; see `shouldFilterOutPathName` for the
carve-outs that apply
+ * regardless of the regex.
+ */
+ val DEFAULT_IGNORED_PATH_SEGMENT_REGEX = "^[._]"
+
+ /** The compiled form of [[DEFAULT_IGNORED_PATH_SEGMENT_REGEX]]. */
+ val defaultIgnoredPathSegmentRegexPattern: Pattern =
+ Pattern.compile(DEFAULT_IGNORED_PATH_SEGMENT_REGEX)
+
/**
* Lists a collection of paths recursively. Picks the listing strategy
adaptively depending
* on the number of paths to list.
@@ -49,6 +61,10 @@ private[spark] object HadoopFSUtils extends Logging {
* @param filter Path filter used to exclude leaf files from result
* @param ignoreMissingFiles Ignore missing files that happen during
recursive listing
* (e.g., due to race conditions)
+ * @param ignoredPathSegmentRegex Regex evaluated with find semantics
against each directory and
+ * file name below the input paths; matching names
are skipped and a
+ * matching directory name excludes its subtree.
See
+ * `shouldFilterOutPathName` for the carve-outs
that always apply.
* @param ignoreLocality Whether to fetch data locality info when listing
leaf files. If false,
* this will return `FileStatus` without
`BlockLocation` info.
* @param parallelismThreshold The threshold to enable parallelism. If the
number of input paths
@@ -65,11 +81,13 @@ private[spark] object HadoopFSUtils extends Logging {
hadoopConf: Configuration,
filter: PathFilter,
ignoreMissingFiles: Boolean,
+ ignoredPathSegmentRegex: Pattern,
ignoreLocality: Boolean,
parallelismThreshold: Int,
parallelismMax: Int): Seq[(Path, Seq[FileStatus])] = {
parallelListLeafFilesInternal(sc, paths, hadoopConf, filter, isRootLevel =
true,
- ignoreMissingFiles, ignoreLocality, parallelismThreshold, parallelismMax)
+ ignoreMissingFiles, ignoredPathSegmentRegex, ignoreLocality,
parallelismThreshold,
+ parallelismMax)
}
/**
@@ -81,12 +99,17 @@ private[spark] object HadoopFSUtils extends Logging {
* @param path a path to list
* @param hadoopConf Hadoop configuration
* @param filter Path filter used to exclude leaf files from result
+ * @param ignoredPathSegmentRegex Regex evaluated with find semantics
against each directory and
+ * file name below `path`; matching names are
skipped. See
+ * `shouldFilterOutPathName` for the carve-outs
that always apply.
* @return the set of discovered files for the path
*/
def listFiles(
path: Path,
hadoopConf: Configuration,
- filter: PathFilter): Seq[(Path, Seq[FileStatus])] = {
+ filter: PathFilter,
+ ignoredPathSegmentRegex: Pattern = defaultIgnoredPathSegmentRegexPattern)
+ : Seq[(Path, Seq[FileStatus])] = {
logInfo(log"Listing ${MDC(PATH, path)} with listFiles API")
try {
val prefixLength = path.toString.length
@@ -94,7 +117,9 @@ private[spark] object HadoopFSUtils extends Logging {
val statues = new Iterator[LocatedFileStatus]() {
def next(): LocatedFileStatus = remoteIter.next
def hasNext: Boolean = remoteIter.hasNext
- }.filterNot(status =>
shouldFilterOutPath(status.getPath.toString.substring(prefixLength)))
+ }.filterNot(status =>
+ shouldFilterOutPath(
+ status.getPath.toString.substring(prefixLength),
ignoredPathSegmentRegex))
.filter(f => filter.accept(f.getPath))
.toArray
Seq((path, statues.toImmutableArraySeq))
@@ -113,6 +138,7 @@ private[spark] object HadoopFSUtils extends Logging {
filter: PathFilter,
isRootLevel: Boolean,
ignoreMissingFiles: Boolean,
+ ignoredPathSegmentRegex: Pattern,
ignoreLocality: Boolean,
parallelismThreshold: Int,
parallelismMax: Int): Seq[(Path, Seq[FileStatus])] = {
@@ -126,6 +152,7 @@ private[spark] object HadoopFSUtils extends Logging {
filter,
Some(sc),
ignoreMissingFiles = ignoreMissingFiles,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex,
ignoreLocality = ignoreLocality,
isRootPath = isRootLevel,
parallelismThreshold = parallelismThreshold,
@@ -166,6 +193,7 @@ private[spark] object HadoopFSUtils extends Logging {
filter = filter,
contextOpt = None, // Can't execute parallel scans on workers
ignoreMissingFiles = ignoreMissingFiles,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex,
ignoreLocality = ignoreLocality,
isRootPath = isRootLevel,
parallelismThreshold = Int.MaxValue,
@@ -193,6 +221,7 @@ private[spark] object HadoopFSUtils extends Logging {
filter: PathFilter,
contextOpt: Option[SparkContext],
ignoreMissingFiles: Boolean,
+ ignoredPathSegmentRegex: Pattern,
ignoreLocality: Boolean,
isRootPath: Boolean,
parallelismThreshold: Int,
@@ -251,7 +280,8 @@ private[spark] object HadoopFSUtils extends Logging {
}
val filteredStatuses =
- statuses.filterNot(status =>
shouldFilterOutPathName(status.getPath.getName))
+ statuses.filterNot(status =>
+ shouldFilterOutPathName(status.getPath.getName,
ignoredPathSegmentRegex))
val allLeafStatuses = {
val (dirs, topLevelFiles) = filteredStatuses.partition(_.isDirectory)
@@ -264,6 +294,7 @@ private[spark] object HadoopFSUtils extends Logging {
filter = filter,
isRootLevel = false,
ignoreMissingFiles = ignoreMissingFiles,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex,
ignoreLocality = ignoreLocality,
parallelismThreshold = parallelismThreshold,
parallelismMax = parallelismMax
@@ -276,6 +307,7 @@ private[spark] object HadoopFSUtils extends Logging {
filter = filter,
contextOpt = contextOpt,
ignoreMissingFiles = ignoreMissingFiles,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex,
ignoreLocality = ignoreLocality,
isRootPath = false,
parallelismThreshold = parallelismThreshold,
@@ -342,39 +374,35 @@ private[spark] object HadoopFSUtils extends Logging {
}
// scalastyle:on argcount
- /** Checks if we should filter out this path name. */
- def shouldFilterOutPathName(pathName: String): Boolean = {
- // We filter follow paths:
- // 1. everything that starts with _ and ., except _common_metadata and
_metadata
- // because Parquet needs to find those metadata files from leaf files
returned by this method.
+ /**
+ * Checks if we should filter out this path name, i.e. whether
`ignoredPathSegmentRegex` finds a
+ * match in it. Regardless of the regex, names starting with
`_common_metadata` or `_metadata`
+ * are never filtered, names ending with `._COPYING_` are always filtered,
and '_'-prefixed
+ * names containing '=' (partition directories) are never filtered.
+ */
+ def shouldFilterOutPathName(
+ pathName: String,
+ ignoredPathSegmentRegex: Pattern =
defaultIgnoredPathSegmentRegexPattern): Boolean = {
+ // Parquet summary files must stay visible to listing regardless of the
filter, because
+ // Parquet needs to find those metadata files from leaf files returned by
this method.
// We should refactor this logic to not mix metadata files with data files.
- // 2. everything that ends with `._COPYING_`, because this is a
intermediate state of file. we
- // should skip this file in case of double reading.
- val exclude = (pathName.startsWith("_") && !pathName.contains("=")) ||
- pathName.startsWith(".") || pathName.endsWith("._COPYING_")
- val include = pathName.startsWith("_common_metadata") ||
pathName.startsWith("_metadata")
- exclude && !include
+ if (pathName.startsWith("_common_metadata") ||
pathName.startsWith("_metadata")) return false
+ // In-flight copy files (hadoop fs -put) are always skipped to avoid
double reads.
+ if (pathName.endsWith("._COPYING_")) return true
+ // '_'-prefixed names containing '=' are partition directories, never
hidden.
+ if (pathName.startsWith("_") && pathName.contains("=")) return false
+ ignoredPathSegmentRegex.matcher(pathName).find()
}
- private val underscore: Regex = "/_[^=/]*/".r
- private val underscoreEnd: Regex = "/_[^=/]*$".r
-
- /** Checks if we should filter out this path. */
- @scala.annotation.tailrec
- def shouldFilterOutPath(path: String): Boolean = {
- if (path.contains("/.") || path.endsWith("._COPYING_")) return true
- underscoreEnd.findFirstIn(path) match {
- case Some(dir) if dir.equals("/_metadata") ||
dir.equals("/_common_metadata") => false
- case Some(_) => true
- case None =>
- underscore.findFirstIn(path) match {
- case Some(dir) if dir.equals("/_metadata/") =>
- shouldFilterOutPath(path.replaceFirst("/_metadata", ""))
- case Some(dir) if dir.equals("/_common_metadata/") =>
- shouldFilterOutPath(path.replaceFirst("/_common_metadata", ""))
- case Some(_) => true
- case None => false
- }
- }
+ /**
+ * Checks if we should filter out this path, i.e. whether any of its
individual directory or
+ * file name components below the listed base path is filtered by
`ignoredPathSegmentRegex`
+ * according to [[shouldFilterOutPathName]].
+ */
+ def shouldFilterOutPath(
+ path: String,
+ ignoredPathSegmentRegex: Pattern =
defaultIgnoredPathSegmentRegexPattern): Boolean = {
+ path.split("/").exists(name =>
+ name.nonEmpty && shouldFilterOutPathName(name, ignoredPathSegmentRegex))
}
}
diff --git a/core/src/test/scala/org/apache/spark/util/HadoopFSUtilsSuite.scala
b/core/src/test/scala/org/apache/spark/util/HadoopFSUtilsSuite.scala
index e715d9fb293a..1ee96b90e3cb 100644
--- a/core/src/test/scala/org/apache/spark/util/HadoopFSUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/HadoopFSUtilsSuite.scala
@@ -17,9 +17,52 @@
package org.apache.spark.util
-import org.apache.spark.SparkFunSuite
+import java.io.File
+import java.nio.file.Files
+import java.util.regex.Pattern
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileStatus, Path, PathFilter}
+
+import org.apache.spark.{SparkContext, SparkFunSuite}
+import org.apache.spark.LocalSparkContext.withSpark
class HadoopFSUtilsSuite extends SparkFunSuite {
+
+ // Accept everything; hidden-file filtering is exercised via the
ignoredPathSegmentRegex regex.
+ private val acceptAllFilter: PathFilter = AcceptAllPathFilter
+
+ // Never matches any name: disables generic hidden-file filtering, leaving
only the carve-outs.
+ private val neverMatch: Pattern = Pattern.compile("(?!)")
+
+ /**
+ * Builds a tree with one regular file, hidden entries ('_'-, '.'-,
'._COPYING_'-named) and a
+ * hidden subdir with its own file.
+ *
+ * @return a tuple of (root path to pass to the listing APIs, name of the
only non-hidden file
+ * in the tree).
+ */
+ private def createHiddenFileTree(root: File): (Path, String) = {
+ def writeFile(parent: File, name: String): Unit = {
+ val file = new File(parent, name)
+ Files.write(file.toPath, "content".getBytes)
+ }
+ writeFile(root, "data.parquet")
+ writeFile(root, "_hidden")
+ writeFile(root, ".dot")
+ writeFile(root, "x._COPYING_")
+ val hiddenDir = new File(root, "_tmp")
+ assert(hiddenDir.mkdir())
+ writeFile(hiddenDir, "nested.parquet")
+ // Use getCanonicalPath, not toURI: toURI's trailing slash breaks
HadoopFSUtils' prefix
+ // stripping and defeats shouldFilterOutPath's leading-'/' match.
+ (new Path(root.getCanonicalPath), "data.parquet")
+ }
+
+ // The set of leaf-file names surfaced for the root path.
+ private def leafFileNames(listing: Seq[(Path, Seq[FileStatus])]):
Set[String] =
+ listing.flatMap(_._2).map(_.getPath.getName).toSet
+
test("HadoopFSUtils - file filtering") {
assert(!HadoopFSUtils.shouldFilterOutPathName("abcd"))
assert(HadoopFSUtils.shouldFilterOutPathName(".ab"))
@@ -29,6 +72,30 @@ class HadoopFSUtilsSuite extends SparkFunSuite {
assert(HadoopFSUtils.shouldFilterOutPathName("_ab_metadata"))
assert(HadoopFSUtils.shouldFilterOutPathName("_cd_common_metadata"))
assert(HadoopFSUtils.shouldFilterOutPathName("a._COPYING_"))
+ // A never-matching regex surfaces generically hidden names...
+ assert(!HadoopFSUtils.shouldFilterOutPathName(".ab", neverMatch))
+ assert(!HadoopFSUtils.shouldFilterOutPathName("_cd", neverMatch))
+ assert(!HadoopFSUtils.shouldFilterOutPath("/.ab", neverMatch))
+ // ... but the hardcoded carve-outs are not overridable: '._COPYING_'
files stay hidden and
+ // metadata-prefixed files stay visible regardless of the regex.
+ assert(HadoopFSUtils.shouldFilterOutPathName("a._COPYING_", neverMatch))
+ assert(!HadoopFSUtils.shouldFilterOutPathName("_metadata", neverMatch))
+ }
+
+ test("HadoopFSUtils - file filtering with a custom regex") {
+ val backup = Pattern.compile("^backup")
+ // The custom regex hides matching names.
+ assert(HadoopFSUtils.shouldFilterOutPathName("backup_1", backup))
+ // Names not matching the custom regex are listed, even '_'- or
'.'-prefixed ones: the
+ // default '^[._]' regex is replaced, not combined, and no carve-out hides
them.
+ assert(!HadoopFSUtils.shouldFilterOutPathName("_hidden", backup))
+ assert(!HadoopFSUtils.shouldFilterOutPathName(".dot", backup))
+ assert(!HadoopFSUtils.shouldFilterOutPathName("abcd", backup))
+ // The carve-outs still apply around the custom regex: '._COPYING_' names
stay hidden, and
+ // '_'-prefixed names containing '=' (partition dirs) stay visible even
when the regex
+ // finds a match (note find semantics: unanchored "backup" matches inside
"_backup=1").
+ assert(HadoopFSUtils.shouldFilterOutPathName("backup_1._COPYING_", backup))
+ assert(!HadoopFSUtils.shouldFilterOutPathName("_backup=1",
Pattern.compile("backup")))
}
test("SPARK-45452: HadoopFSUtils - path filtering") {
@@ -61,5 +128,68 @@ class HadoopFSUtilsSuite extends SparkFunSuite {
assert(HadoopFSUtils.shouldFilterOutPath("/ab/_cd/part=1"))
assert(HadoopFSUtils.shouldFilterOutPath("/_ab_metadata"))
assert(HadoopFSUtils.shouldFilterOutPath("/_cd_common_metadata"))
+ // Case 4: the per-name-component walk unifies this predicate with the
recursive-descent
+ // listing path. The verdicts below intentionally differ from the old
path-based logic:
+ // a '_metadata' leaf no longer rescues a hidden parent directory...
+ assert(HadoopFSUtils.shouldFilterOutPath("/_foo/_metadata"))
+ // ... the metadata exemption is prefix-based, matching
shouldFilterOutPathName...
+ assert(!HadoopFSUtils.shouldFilterOutPath("/x/_metadata.json"))
+ // ... and a '._COPYING_' directory component hides its subtree.
+ assert(HadoopFSUtils.shouldFilterOutPath("/d._COPYING_/x"))
}
+
+ test("listFiles - default ignoredPathSegmentRegex hides hidden files and
dirs") {
+ withTempDir { root =>
+ val (path, regularFile) = createHiddenFileTree(root)
+ val hadoopConf = new Configuration()
+ val names = leafFileNames(
+ HadoopFSUtils.listFiles(path, hadoopConf, acceptAllFilter))
+ // Only the regular file survives; every hidden entry is filtered out.
+ assert(names === Set(regularFile))
+ }
+ }
+
+ test("listFiles - never-matching ignoredPathSegmentRegex surfaces hidden
files and dirs") {
+ withTempDir { root =>
+ val (path, regularFile) = createHiddenFileTree(root)
+ val hadoopConf = new Configuration()
+ val names = leafFileNames(
+ HadoopFSUtils.listFiles(path, hadoopConf, acceptAllFilter, neverMatch))
+ // 'x._COPYING_' stays hidden even with a never-matching regex: the
carve-out for
+ // in-flight copy files is not overridable.
+ assert(names === Set(regularFile, "_hidden", ".dot", "nested.parquet"))
+ }
+ }
+
+ test("parallelListLeafFiles - ignoredPathSegmentRegex toggles hidden file
visibility") {
+ withTempDir { root =>
+ val (path, regularFile) = createHiddenFileTree(root)
+ val hadoopConf = new Configuration()
+ withSpark(new SparkContext("local", "HadoopFSUtilsSuite")) { sc =>
+ def listNames(ignoredPathSegmentRegex: Pattern): Set[String] =
+ leafFileNames(HadoopFSUtils.parallelListLeafFiles(
+ sc,
+ Seq(path),
+ hadoopConf,
+ acceptAllFilter,
+ ignoreMissingFiles = false,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex,
+ ignoreLocality = true,
+ // Use 0 so the parallel (Spark job) code path is exercised rather
than the
+ // serial short-circuit.
+ parallelismThreshold = 0,
+ parallelismMax = 1))
+
+ val defaultFilter =
Pattern.compile(HadoopFSUtils.DEFAULT_IGNORED_PATH_SEGMENT_REGEX)
+ assert(listNames(defaultFilter) === Set(regularFile))
+ // 'x._COPYING_' stays hidden even with a never-matching regex (see
the listFiles test).
+ assert(listNames(neverMatch) ===
+ Set(regularFile, "_hidden", ".dot", "nested.parquet"))
+ }
+ }
+ }
+}
+
+private object AcceptAllPathFilter extends PathFilter with Serializable {
+ override def accept(path: Path): Boolean = true
}
diff --git a/docs/sql-data-sources-generic-options.md
b/docs/sql-data-sources-generic-options.md
index d180f15de27a..f1066648cc71 100644
--- a/docs/sql-data-sources-generic-options.md
+++ b/docs/sql-data-sources-generic-options.md
@@ -97,6 +97,46 @@ you can use:
</div>
</div>
+### Ignored Path Segment Regex
+
+Spark allows you to use the configuration
`spark.sql.files.ignoredPathSegmentRegex` or the data source option
`ignoredPathSegmentRegex` to control which files are treated as
+hidden during file listing. The value is a Java regular expression that is
matched (with find semantics, i.e. `java.util.regex.Matcher.find`) against each
individual
+directory and file name below the path being read; names in which the regex
finds a match are skipped from file listing, partition discovery, and reads,
and a matching
+directory name excludes its whole subtree. The default value is `^[._]`, which
skips files and directories whose names start with `_` or `.`. The data source
option
+takes precedence over the configuration when both are set.
+
+Regardless of the regex, three rules always apply: names starting with
`_metadata` or `_common_metadata` (Parquet summary files) are always listed,
names ending in
+`._COPYING_` (in-flight copies) are always skipped, and `_`-prefixed names
containing `=` (partition directories) are always kept.
+
+An empty string disables the generic hidden-file filtering (an empty regex
matches nothing) and surfaces hidden files, including Spark-internal marker
files such as
+`_SUCCESS` and files under `_temporary` directories (the three rules above
still apply). Note the difference from `pathGlobFilter`: `pathGlobFilter` is an
+include-style glob applied to leaf file names only, while
`ignoredPathSegmentRegex` is an exclude-style regex applied to every directory
and file name component; the two
+can be combined, e.g. using `pathGlobFilter` to narrow the results of a
relaxed `ignoredPathSegmentRegex`.
+
+Note that directories surfaced by a relaxed regex also participate in
partition discovery, so a hidden directory next to partition directories causes
a conflicting
+directory structures error unless
`spark.sql.files.ignoreInvalidPartitionPaths` is enabled.
+
+To surface files that are hidden by default, you can use:
+
+<div class="codetabs">
+
+<div data-lang="python" markdown="1">
+{% include_example ignored_path_segment_regex python/sql/datasource.py %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% include_example ignored_path_segment_regex
scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala %}
+</div>
+
+<div data-lang="java" markdown="1">
+{% include_example ignored_path_segment_regex
java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java %}
+</div>
+
+<div data-lang="r" markdown="1">
+{% include_example ignored_path_segment_regex r/RSparkSQLExample.R %}
+</div>
+</div>
+
### Recursive File Lookup
`recursiveFileLookup` is used to recursively load files and it disables
partition inferring. Its default value is `false`.
If data source explicitly specifies the `partitionSpec` when
`recursiveFileLookup` is true, exception will be thrown.
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 1ad7c97ef5e0..460a4d4219bf 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -24,6 +24,7 @@ license: |
## Upgrading from Spark SQL 4.2 to 4.3
+- Since Spark 4.3, zero-length files are skipped during Parquet schema
inference instead of failing with a `FAILED_READ_FILE.CANNOT_READ_FILE_FOOTER`
error.
- Since Spark 4.3, the configuration key
`spark.sql.sources.v2.bucketing.allowJoinKeysSubsetOfPartitionKeys.enabled` has
been renamed to
`spark.sql.sources.v2.bucketing.allowKeysSubsetOfPartitionKeys.enabled` to
reflect that it now applies to storage-partitioned joins, aggregates, and
windows. The old key continues to work as an alias.
- Since Spark 4.3, the Spark Thrift Server rejects setting JVM system
properties through the `set:system:` session configuration overlay (for
example, in a JDBC connection string). To restore the previous behavior, set
`spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties` to `true`.
diff --git a/docs/streaming/apis-on-dataframes-and-datasets.md
b/docs/streaming/apis-on-dataframes-and-datasets.md
index 85815b54d426..86585caead51 100644
--- a/docs/streaming/apis-on-dataframes-and-datasets.md
+++ b/docs/streaming/apis-on-dataframes-and-datasets.md
@@ -75,6 +75,8 @@ Here are the details of all the sources in Spark.
"s3n://a/b/dataset.txt"<br/>
"s3a://a/b/c/dataset.txt"
<br/>
+ <code>ignoredPathSegmentRegex</code>: a Java regular expression
matched against each directory and file name during file listing; matching
names are treated as hidden and skipped (default: "^[._]", which skips names
starting with '_' or '.')
+ <br/>
<code>maxFileAge</code>: Maximum age of a file that can be found in
this directory, before it is ignored. For the first batch all files will be
considered valid. If <code>latestFirst</code> is set to `true` and
<code>maxFilesPerTrigger</code> or <code>maxBytesPerTrigger</code> is set, then
this parameter will be ignored, because old files that are valid, and should be
processed, may be ignored. The max age is specified with respect to the
timestamp of the latest file, and not the [...]
<br/>
<code>maxCachedFiles</code>: maximum number of files to cache to be
processed in subsequent batches (default: 10000). If files are available in
the cache, they will be read from first before listing from the input source.
diff --git
a/examples/src/main/java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java
b/examples/src/main/java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java
index e7d52ce03e86..3dcb0ff05001 100644
---
a/examples/src/main/java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java
+++
b/examples/src/main/java/org/apache/spark/examples/sql/JavaSQLDataSourceExample.java
@@ -156,6 +156,13 @@ public class JavaSQLDataSourceExample {
// |file2.parquet|
// +-------------+
// $example off:recursive_file_lookup$
+ // $example on:ignored_path_segment_regex$
+ // An empty regex surfaces files hidden by default (e.g. names starting
with "_" or ".")
+ Dataset<Row> surfacedDF = spark.read().format("parquet")
+ .option("ignoredPathSegmentRegex", "")
+ .load("examples/src/main/resources/dir1");
+ surfacedDF.show();
+ // $example off:ignored_path_segment_regex$
spark.sql("set spark.sql.files.ignoreCorruptFiles=false");
// $example on:load_with_path_glob_filter$
Dataset<Row> testGlobFilterDF = spark.read().format("parquet")
diff --git a/examples/src/main/python/sql/datasource.py
b/examples/src/main/python/sql/datasource.py
index 8747bd11f8f2..0a76376e2c81 100644
--- a/examples/src/main/python/sql/datasource.py
+++ b/examples/src/main/python/sql/datasource.py
@@ -67,6 +67,14 @@ def generic_file_source_options_example(spark: SparkSession)
-> None:
# |file2.parquet|
# +-------------+
# $example off:recursive_file_lookup$
+
+ # $example on:ignored_path_segment_regex$
+ # An empty regex surfaces files that are hidden by default (e.g. names
starting with "_" or ".")
+ surfaced_df = spark.read.format("parquet")\
+ .option("ignoredPathSegmentRegex", "")\
+ .load("examples/src/main/resources/dir1")
+ surfaced_df.show()
+ # $example off:ignored_path_segment_regex$
spark.sql("set spark.sql.files.ignoreCorruptFiles=false")
# $example on:load_with_path_glob_filter$
diff --git a/examples/src/main/r/RSparkSQLExample.R
b/examples/src/main/r/RSparkSQLExample.R
index a7d3ae766c5a..15e9c8174659 100644
--- a/examples/src/main/r/RSparkSQLExample.R
+++ b/examples/src/main/r/RSparkSQLExample.R
@@ -126,6 +126,12 @@ head(recursiveLoadedDF)
# 1 file1.parquet
# 2 file2.parquet
# $example off:recursive_file_lookup$
+
+# $example on:ignored_path_segment_regex$
+# An empty regex surfaces files that are hidden by default (e.g. names
starting with "_" or ".")
+surfacedDF <- read.df("examples/src/main/resources/dir1", "parquet",
ignoredPathSegmentRegex = "")
+head(surfacedDF)
+# $example off:ignored_path_segment_regex$
sql("set spark.sql.files.ignoreCorruptFiles=false")
# $example on:generic_load_save_functions$
diff --git
a/examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala
b/examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala
index 33121586fe10..a6282ab230a0 100644
---
a/examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala
+++
b/examples/src/main/scala/org/apache/spark/examples/sql/SQLDataSourceExample.scala
@@ -85,6 +85,13 @@ object SQLDataSourceExample {
// |file2.parquet|
// +-------------+
// $example off:recursive_file_lookup$
+ // $example on:ignored_path_segment_regex$
+ // An empty regex surfaces files hidden by default (e.g. names starting
with "_" or ".")
+ val surfacedDF = spark.read.format("parquet")
+ .option("ignoredPathSegmentRegex", "")
+ .load("examples/src/main/resources/dir1")
+ surfacedDF.show()
+ // $example off:ignored_path_segment_regex$
spark.sql("set spark.sql.files.ignoreCorruptFiles=false")
// $example on:load_with_path_glob_filter$
val testGlobFilterDF = spark.read.format("parquet")
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala
index 0747a8045e7d..62db2d90ec8f 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala
@@ -16,7 +16,9 @@
*/
package org.apache.spark.sql.catalyst
-import org.apache.spark.sql.catalyst.FileSourceOptions.{IGNORE_CORRUPT_FILES,
IGNORE_MISSING_FILES}
+import java.util.regex.{Pattern, PatternSyntaxException}
+
+import org.apache.spark.sql.catalyst.FileSourceOptions.{IGNORE_CORRUPT_FILES,
IGNORE_MISSING_FILES, IGNORED_PATH_SEGMENT_REGEX}
import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateFormatter}
import org.apache.spark.sql.internal.{LegacyBehaviorPolicy, SQLConf}
@@ -53,9 +55,46 @@ class FileSourceOptions(
* executors. Only the CSV data source currently honors this.
*/
val archiveFormatEnabled: Boolean =
SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+
+ val ignoredPathSegmentRegex: String =
+ parameters.getOrElse(IGNORED_PATH_SEGMENT_REGEX,
SQLConf.get.ignoredPathSegmentRegex)
+
+ /**
+ * The effective [[ignoredPathSegmentRegex]] compiled and validated once
here, so the ~5 listing
+ * call sites can reuse a single [[Pattern]] instead of re-compiling. An
empty value disables the
+ * generic hidden-file filter (see
[[FileSourceOptions.compileIgnoredPathSegmentRegex]]).
+ */
+ lazy val ignoredPathSegmentRegexPattern: Pattern =
+ FileSourceOptions.compileIgnoredPathSegmentRegex(ignoredPathSegmentRegex)
}
object FileSourceOptions {
val IGNORE_CORRUPT_FILES = "ignoreCorruptFiles"
val IGNORE_MISSING_FILES = "ignoreMissingFiles"
+ val IGNORED_PATH_SEGMENT_REGEX = "ignoredPathSegmentRegex"
+
+ // A regex that never matches any name, used when the filter is disabled by
an empty value.
+ private val DISABLED_FILTER_PATTERN = Pattern.compile("(?!)")
+
+ /**
+ * Compiles the effective `ignoredPathSegmentRegex`. An empty string
disables the generic
+ * hidden-file filter (logically an "empty" regex matches nothing, so the
returned pattern never
+ * matches). A non-empty value must be a valid Java regular expression; an
invalid one is reported
+ * as a clear [[IllegalArgumentException]] rather than a raw
[[PatternSyntaxException]] surfacing
+ * deep in file listing.
+ */
+ def compileIgnoredPathSegmentRegex(regex: String): Pattern = {
+ if (regex.isEmpty) {
+ DISABLED_FILTER_PATTERN
+ } else {
+ try {
+ Pattern.compile(regex)
+ } catch {
+ case e: PatternSyntaxException =>
+ throw new IllegalArgumentException(
+ s"The '$IGNORED_PATH_SEGMENT_REGEX' value '$regex' is not a valid
Java regular " +
+ "expression. Use an empty string to disable hidden-file
filtering.", e)
+ }
+ }
+ }
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 05043b3107bf..bda3f1cf1772 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -21,6 +21,7 @@ import java.util.{Locale, Properties, TimeZone}
import java.util
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicReference
+import java.util.regex.Pattern
import scala.collection.immutable
import scala.jdk.CollectionConverters._
@@ -52,7 +53,7 @@ import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors
import org.apache.spark.sql.types.{AtomicType, TimestampNTZType, TimestampType}
import org.apache.spark.storage.{StorageLevel, StorageLevelMapper}
import org.apache.spark.unsafe.array.ByteArrayMethods
-import org.apache.spark.util.{Utils, VersionUtils}
+import org.apache.spark.util.{HadoopFSUtils, Utils, VersionUtils}
////////////////////////////////////////////////////////////////////////////////////////////////////
// This file defines the configuration options for Spark SQL.
@@ -2785,6 +2786,27 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val IGNORED_PATH_SEGMENT_REGEX =
buildConf("spark.sql.files.ignoredPathSegmentRegex")
+ .doc("Java regular expression matched (with find semantics) against each
directory and " +
+ "file name during file listing; matching names are skipped from listing,
partition " +
+ "discovery, and reads. The default '^[._]' skips names starting with '_'
or '.' (hidden " +
+ "files). Regardless of the regex, names starting with '_metadata' or
'_common_metadata' " +
+ "are always listed, names ending in '._COPYING_' are always skipped, and
'_'-prefixed " +
+ "names containing '=' (partition directories) are always kept. This
configuration is " +
+ "effective only when using file-based sources such as Parquet, JSON and
ORC. It can be " +
+ "overridden per read by the 'ignoredPathSegmentRegex' data source
option. An empty string " +
+ "disables generic hidden-file filtering (an empty regex matches
nothing); note that " +
+ "directories it surfaces also participate in partition discovery, so a
hidden directory " +
+ "next to partition directories causes a conflicting directory structures
error unless " +
+ "'spark.sql.files.ignoreInvalidPartitionPaths' is enabled.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .stringConf
+ .checkValue(v => v.isEmpty || Try(Pattern.compile(v)).isSuccess,
+ "The value of spark.sql.files.ignoredPathSegmentRegex must be empty (to
disable " +
+ "hidden-file filtering) or a valid Java regular expression.")
+ .createWithDefault(HadoopFSUtils.DEFAULT_IGNORED_PATH_SEGMENT_REGEX)
+
val IGNORE_INVALID_PARTITION_PATHS =
buildConf("spark.sql.files.ignoreInvalidPartitionPaths")
.doc("Whether to ignore invalid partition paths that do not match
<column>=<value>. When " +
"the option is enabled, table with two partition directories
'table/invalid' and " +
@@ -7883,6 +7905,8 @@ class SQLConf extends Serializable with Logging with
SqlApiConf {
def ignoreMissingFiles: Boolean = getConf(IGNORE_MISSING_FILES)
+ def ignoredPathSegmentRegex: String = getConf(IGNORED_PATH_SEGMENT_REGEX)
+
def ignoreInvalidPartitionPaths: Boolean =
getConf(IGNORE_INVALID_PARTITION_PATHS)
def maxRecordsPerFile: Long = getConf(MAX_RECORDS_PER_FILE)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
index b9ff24139f07..3eb2b14539ff 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
@@ -26,7 +26,7 @@ import org.apache.hadoop.fs.{FileStatus, FileSystem, Path,
PathFilter}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{COUNT, DATABASE_NAME, ERROR,
TABLE_NAME, TIME}
-import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
+import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow,
TableIdentifier}
import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
import org.apache.spark.sql.catalyst.catalog.{CatalogStatistics, CatalogTable,
CatalogTablePartition, ExternalCatalogUtils}
@@ -44,6 +44,7 @@ import
org.apache.spark.sql.execution.datasources.v2.ExtractV2CatalogAndIdentifi
import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.internal.{SessionState, SQLConf}
import org.apache.spark.sql.types._
+import org.apache.spark.util.HadoopFSUtils
import org.apache.spark.util.collection.Utils
/**
@@ -197,8 +198,14 @@ object CommandUtils extends Logging {
val stagingDir = sparkSession.sessionState.conf
.getConfString("hive.exec.stagingdir", ".hive-staging")
val filter = new PathFilterIgnoreNonData(stagingDir)
+ // Pin the default ignoredPathSegmentRegex: stats must always skip hidden
dirs (immune to the
+ // session conf), matching calculateSingleLocationSize.
val sizes = InMemoryFileIndex.bulkListLeafFiles(paths.flatten,
- sparkSession.sessionState.newHadoopConf(), filter, sparkSession).map {
+ sparkSession.sessionState.newHadoopConf(), filter, sparkSession,
+ parameters = Map(
+ FileSourceOptions.IGNORED_PATH_SEGMENT_REGEX ->
+ HadoopFSUtils.DEFAULT_IGNORED_PATH_SEGMENT_REGEX)
+ ).map {
case (_, files) => files.map(_.getLen).sum
}
// the size is 0 where paths(i) is not defined and sizes(i) where it is
defined
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
index c0b34fd9ef9e..84daeb92fe8e 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources
import java.io.{Closeable, InputStream}
import java.util.Locale
+import java.util.regex.Pattern
import java.util.zip.GZIPInputStream
import scala.util.control.NonFatal
@@ -59,9 +60,13 @@ abstract class ArchiveReader(path: Path) {
* only once the current entry's iterator is exhausted, so nothing is
buffered to disk and at most
* one entry's bytes are read at a time. The archive stream is closed when
the returned iterator
* is exhausted, when [[Closeable.close]] is called on it, and (defensively)
on task completion.
+ * Entry skipping mirrors Spark's file listing: `ignoredPathSegmentRegex` is
the effective filter
+ * (the `ignoredPathSegmentRegex` data source option), so callers reading
with a custom filter
+ * must pass its compiled form here for archive entries to be filtered like
loose files.
*/
def readEntries[T](
- conf: Configuration)(
+ conf: Configuration,
+ ignoredPathSegmentRegex: Pattern =
HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)(
parseEntry: (String, InputStream) => Iterator[T]): Iterator[T]
}
@@ -147,16 +152,19 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
/**
* Whether an entry is not a real data file and must be skipped: a
directory, or a name Spark's
* own file listing would filter out. Applying
[[HadoopFSUtils.shouldFilterOutPathName]] (the
- * `InMemoryFileIndex` filter) to every path component keeps archive reads
in parity with reading
- * the same entries as loose files: `.`-prefixed sidecars (macOS `._x`,
`.DS_Store`), `_`-prefixed
- * markers (`_SUCCESS`, `_committed_*`), and anything under a
`.`/`_`-prefixed directory (e.g. a
- * leftover `_temporary/` from a failed write) are skipped, while data files
are kept. The
- * per-component check matters because `InMemoryFileIndex` never recurses
into such directories,
- * so a basename-only filter would read `_temporary/part-0.csv` that a
loose-file scan drops.
+ * `InMemoryFileIndex` filter) with the same effective
`ignoredPathSegmentRegex` to every path
+ * component keeps archive reads in parity with reading the same entries as
loose files,
+ * including when the user supplies a custom `ignoredPathSegmentRegex`
option: under the default
+ * filter, `.`-prefixed sidecars (macOS `._x`, `.DS_Store`), `_`-prefixed
markers (`_SUCCESS`,
+ * `_committed_*`), and anything under a `.`/`_`-prefixed directory (e.g. a
leftover
+ * `_temporary/` from a failed write) are skipped, while data files are
kept. The per-component
+ * check matters because `InMemoryFileIndex` never recurses into such
directories, so a
+ * basename-only filter would read `_temporary/part-0.csv` that a loose-file
scan drops.
*/
- private def shouldSkipEntry(entry: TarArchiveEntry): Boolean = {
+ private def shouldSkipEntry(entry: TarArchiveEntry, ignoredPathSegmentRegex:
Pattern): Boolean = {
if (entry.isDirectory) return true
- entry.getName.split("/").exists(c => c.nonEmpty &&
HadoopFSUtils.shouldFilterOutPathName(c))
+ entry.getName.split("/").exists(c =>
+ c.nonEmpty && HadoopFSUtils.shouldFilterOutPathName(c,
ignoredPathSegmentRegex))
}
/** Opens the archive as a tar stream, transparently decompressing `.tar.gz`
/ `.tgz`. */
@@ -176,7 +184,8 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
CloseShieldInputStream.wrap(tar)
override def readEntries[T](
- conf: Configuration)(
+ conf: Configuration,
+ ignoredPathSegmentRegex: Pattern)(
parseEntry: (String, InputStream) => Iterator[T]): Iterator[T] = {
val tar = openTarStream(conf)
var closed = false
@@ -207,7 +216,9 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
case _ =>
}
var entry = tar.getNextEntry
- while (entry != null && shouldSkipEntry(entry)) entry =
tar.getNextEntry
+ while (entry != null && shouldSkipEntry(entry,
ignoredPathSegmentRegex)) {
+ entry = tar.getNextEntry
+ }
if (entry == null) {
done = true
cleanup()
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
index 4a95f681fb6e..e249e8cfa4cf 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSource.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.datasources
import java.util.{Locale, ServiceConfigurationError, ServiceLoader}
+import java.util.regex.Pattern
import scala.jdk.CollectionConverters._
import scala.util.{Failure, Success, Try}
@@ -30,7 +31,7 @@ import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{CLASS_NAME, DATA_SOURCE,
DATA_SOURCES, PATHS}
import org.apache.spark.sql.{AnalysisException, SaveMode, SparkSession}
-import org.apache.spark.sql.catalyst.DataSourceOptions
+import org.apache.spark.sql.catalyst.{DataSourceOptions, FileSourceOptions}
import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.catalog.{BucketSpec,
CatalogStorageFormat, CatalogTable, CatalogUtils}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
@@ -589,8 +590,11 @@ case class DataSource(
checkEmptyGlobPath: Boolean,
checkFilesExist: Boolean): Seq[Path] = {
val allPaths = caseInsensitiveOptions.get("path") ++ paths
+ val ignoredPathSegmentRegex =
+ new
FileSourceOptions(caseInsensitiveOptions).ignoredPathSegmentRegexPattern
DataSource.checkAndGlobPathIfNecessary(allPaths.toSeq,
newHadoopConfiguration(),
- checkEmptyGlobPath, checkFilesExist, enableGlobbing = globPaths)
+ checkEmptyGlobPath, checkFilesExist, enableGlobbing = globPaths,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex)
}
private def disallowWritingIntervals(
@@ -797,7 +801,9 @@ object DataSource extends Logging {
checkEmptyGlobPath: Boolean,
checkFilesExist: Boolean,
numThreads: Integer = 40,
- enableGlobbing: Boolean): Seq[Path] = {
+ enableGlobbing: Boolean,
+ ignoredPathSegmentRegex: Pattern =
HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)
+ : Seq[Path] = {
val qualifiedPaths = pathStrings.map { pathString =>
val path = new Path(pathString)
val fs = path.getFileSystem(hadoopConf)
@@ -844,7 +850,7 @@ object DataSource extends Logging {
val allPaths = globbedPaths ++ nonGlobPaths
if (checkFilesExist) {
val (filteredOut, filteredIn) = allPaths.partition { path =>
- HadoopFSUtils.shouldFilterOutPathName(path.getName)
+ HadoopFSUtils.shouldFilterOutPathName(path.getName,
ignoredPathSegmentRegex)
}
if (filteredIn.isEmpty) {
logWarning(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndexOptions.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndexOptions.scala
index 5a300dae4daa..0a7a775b6469 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndexOptions.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FileIndexOptions.scala
@@ -22,6 +22,7 @@ import org.apache.spark.sql.catalyst.util.DateTimeUtils
object FileIndexOptions extends DataSourceOptions {
val IGNORE_MISSING_FILES = newOption(FileSourceOptions.IGNORE_MISSING_FILES)
+ val IGNORED_PATH_SEGMENT_REGEX =
newOption(FileSourceOptions.IGNORED_PATH_SEGMENT_REGEX)
val IGNORE_INVALID_PARTITION_PATHS = newOption("ignoreInvalidPartitionPaths")
val TIME_ZONE = newOption(DateTimeUtils.TIMEZONE_OPTION)
val RECURSIVE_FILE_LOOKUP = newOption("recursiveFileLookup")
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala
index 2d68faa3ff52..05273c2a8f95 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala
@@ -121,8 +121,14 @@ class InMemoryFileIndex(
val startTime = System.nanoTime()
val output = mutable.LinkedHashSet[FileStatus]()
val pathsToFetch = mutable.ArrayBuffer[Path]()
+ // The file status cache is keyed by path only, so cached entries must
always hold the
+ // default-filtered listing. Bypass the cache when a non-default
ignoredPathSegmentRegex is in
+ // effect, to neither serve nor store listings produced with a different
filter.
+ val useFileStatusCache =
+ ignoredPathSegmentRegex ==
HadoopFSUtils.DEFAULT_IGNORED_PATH_SEGMENT_REGEX
for (path <- paths) {
- fileStatusCache.getLeafFiles(path) match {
+ val cachedFiles = if (useFileStatusCache)
fileStatusCache.getLeafFiles(path) else None
+ cachedFiles match {
case Some(files) =>
HiveCatalogMetrics.incrementFileCacheHits(files.length)
output ++= files
@@ -135,7 +141,9 @@ class InMemoryFileIndex(
pathsToFetch.toSeq, hadoopConf, filter, sparkSession, parameters)
discovered.foreach { case (path, leafFiles) =>
HiveCatalogMetrics.incrementFilesDiscovered(leafFiles.size)
- fileStatusCache.putLeafFiles(path, leafFiles.toArray)
+ if (useFileStatusCache) {
+ fileStatusCache.putLeafFiles(path, leafFiles.toArray)
+ }
output ++= leafFiles
}
logInfo(log"It took ${MDC(ELAPSED_TIME, (System.nanoTime() - startTime) /
(1000 * 1000))} ms" +
@@ -154,8 +162,9 @@ object InMemoryFileIndex extends Logging {
parameters: Map[String, String] = Map.empty): Seq[(Path,
Seq[FileStatus])] = {
val fileSystemList =
sparkSession.sessionState.conf.useListFilesFileSystemList.split(",").map(_.trim)
- val ignoreMissingFiles =
- new FileSourceOptions(CaseInsensitiveMap(parameters)).ignoreMissingFiles
+ val fileSourceOptions = new
FileSourceOptions(CaseInsensitiveMap(parameters))
+ val ignoreMissingFiles = fileSourceOptions.ignoreMissingFiles
+ val ignoredPathSegmentRegex =
fileSourceOptions.ignoredPathSegmentRegexPattern
val useListFiles = try {
val scheme = paths.head.getFileSystem(hadoopConf).getScheme
paths.size == 1 && fileSystemList.contains(scheme)
@@ -166,7 +175,8 @@ object InMemoryFileIndex extends Logging {
HadoopFSUtils.listFiles(
path = paths.head,
hadoopConf = hadoopConf,
- filter = new PathFilterWrapper(filter))
+ filter = new PathFilterWrapper(filter),
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex)
} else {
HadoopFSUtils.parallelListLeafFiles(
sc = sparkSession.sparkContext,
@@ -174,6 +184,7 @@ object InMemoryFileIndex extends Logging {
hadoopConf = hadoopConf,
filter = new PathFilterWrapper(filter),
ignoreMissingFiles = ignoreMissingFiles,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex,
ignoreLocality = sparkSession.sessionState.conf.ignoreDataLocality,
parallelismThreshold =
sparkSession.sessionState.conf.parallelPartitionDiscoveryThreshold,
parallelismMax =
sparkSession.sessionState.conf.parallelPartitionDiscoveryParallelism)
@@ -182,8 +193,10 @@ object InMemoryFileIndex extends Logging {
}
+// Delegates to the input-path filter only (with a null guard). The
hidden-file filtering is owned
+// by HadoopFSUtils.listFiles / parallelListLeafFiles via their
`ignoredPathSegmentRegex` argument,
+// which already evaluates the regex against every directory and file name, so
re-applying it here
+// would be redundant.
private class PathFilterWrapper(val filter: PathFilter) extends PathFilter
with Serializable {
- override def accept(path: Path): Boolean = {
- (filter == null || filter.accept(path)) &&
!HadoopFSUtils.shouldFilterOutPathName(path.getName)
- }
+ override def accept(path: Path): Boolean = filter == null ||
filter.accept(path)
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
index 8cea2c95e694..7e6b733e6aaf 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningAwareFileIndex.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.datasources
import java.io.FileNotFoundException
+import java.util.regex.Pattern
import scala.collection.mutable
@@ -28,7 +29,7 @@ import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{COUNT, PERCENT, TOTAL}
import org.apache.spark.paths.SparkPath
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.{expressions, InternalRow}
+import org.apache.spark.sql.catalyst.{expressions, FileSourceOptions,
InternalRow}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -78,6 +79,15 @@ abstract class PartitioningAwareFileIndex(
.getOrElse(sparkSession.sessionState.conf.ignoreInvalidPartitionPaths)
}
+ protected lazy val ignoredPathSegmentRegex: String = {
+ caseInsensitiveMap
+ .get(FileIndexOptions.IGNORED_PATH_SEGMENT_REGEX)
+ .getOrElse(sparkSession.sessionState.conf.ignoredPathSegmentRegex)
+ }
+
+ private lazy val ignoredPathSegmentRegexPattern: Pattern =
+ FileSourceOptions.compileIgnoredPathSegmentRegex(ignoredPathSegmentRegex)
+
override def listFiles(
partitionFilters: Seq[Expression], dataFilters: Seq[Expression]):
Seq[PartitionDirectory] = {
def isNonEmptyFile(f: FileStatus): Boolean = {
@@ -265,6 +275,12 @@ abstract class PartitioningAwareFileIndex(
// counted as data files, so that they shouldn't participate partition
discovery.
private def isDataPath(path: Path): Boolean = {
val name = path.getName
- !((name.startsWith("_") && !name.contains("=")) || name.startsWith("."))
+ // Partition directories ('_'-prefixed names containing '=') are always
data paths.
+ if (name.startsWith("_") && name.contains("=")) return true
+ // Metadata summary files are never data paths, regardless of the
ignoredPathSegmentRegex regex
+ // (they always survive listing via the shouldFilterOutPathName carve-out).
+ if (name.startsWith("_common_metadata") || name.startsWith("_metadata"))
return false
+ // For all other names the ignoredPathSegmentRegex regex decides.
+ !ignoredPathSegmentRegexPattern.matcher(name).find()
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
index 7424d43341b2..10771114bc70 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources.csv
import java.io.{FileNotFoundException, InputStream, IOException}
import java.nio.charset.{Charset, StandardCharsets}
+import java.util.regex.Pattern
import scala.util.control.NonFatal
@@ -119,13 +120,16 @@ abstract class CSVDataSource extends Serializable with
Logging {
*
* @param getParser builds a fresh [[UnivocityParser]].
* @param getHeaderChecker builds a fresh [[CSVHeaderChecker]] for
`(isStartOfFile, source)`.
+ * @param ignoredPathSegmentRegex the compiled effective
`ignoredPathSegmentRegex` option, so
+ * hidden entries are skipped exactly like Spark's
file listing would.
*/
def readArchive(
conf: Configuration,
file: PartitionedFile,
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
- requiredSchema: StructType): Iterator[InternalRow]
+ requiredSchema: StructType,
+ ignoredPathSegmentRegex: Pattern): Iterator[InternalRow]
/**
* Shared driver used by the [[readArchive]] implementations: streams each
non-skipped entry's
@@ -137,10 +141,11 @@ abstract class CSVDataSource extends Serializable with
Logging {
conf: Configuration,
file: PartitionedFile,
getParser: () => UnivocityParser,
- getHeaderChecker: (Boolean, String) => CSVHeaderChecker)(
+ getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
+ ignoredPathSegmentRegex: Pattern)(
parseEntry: (UnivocityParser, CSVHeaderChecker, InputStream) =>
Iterator[InternalRow])
: Iterator[InternalRow] = {
- ArchiveReader(file.toPath).readEntries(conf) { (entryName, in) =>
+ ArchiveReader(file.toPath).readEntries(conf, ignoredPathSegmentRegex) {
(entryName, in) =>
val headerChecker =
getHeaderChecker(true, s"CSV archive entry:
${file.urlEncodedPath}!/$entryName")
val parser = getParser()
@@ -296,12 +301,14 @@ object TextInputCSVDataSource extends CSVDataSource {
file: PartitionedFile,
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
- requiredSchema: StructType): Iterator[InternalRow] =
+ requiredSchema: StructType,
+ ignoredPathSegmentRegex: Pattern): Iterator[InternalRow] =
// Stream each tar entry through the line-based parser, treating the entry
exactly like a
// standalone CSV file (a fresh parser/header checker is built per entry).
- streamArchiveEntries(conf, file, getParser, getHeaderChecker) { (parser,
headerChecker, in) =>
- UnivocityParser.parseIterator(
- entryLines(in, parser.options), parser, headerChecker, requiredSchema)
+ streamArchiveEntries(conf, file, getParser, getHeaderChecker,
ignoredPathSegmentRegex) {
+ (parser, headerChecker, in) =>
+ UnivocityParser.parseIterator(
+ entryLines(in, parser.options), parser, headerChecker,
requiredSchema)
}
/**
@@ -425,11 +432,13 @@ object MultiLineCSVDataSource extends CSVDataSource {
file: PartitionedFile,
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
- requiredSchema: StructType): Iterator[InternalRow] =
+ requiredSchema: StructType,
+ ignoredPathSegmentRegex: Pattern): Iterator[InternalRow] =
// Stream each tar entry whole through the multi-line parser (a fresh
parser/header checker is
// built per entry).
- streamArchiveEntries(conf, file, getParser, getHeaderChecker) { (parser,
headerChecker, in) =>
- UnivocityParser.parseStream(in, parser, headerChecker, requiredSchema)
+ streamArchiveEntries(conf, file, getParser, getHeaderChecker,
ignoredPathSegmentRegex) {
+ (parser, headerChecker, in) =>
+ UnivocityParser.parseStream(in, parser, headerChecker, requiredSchema)
}
override protected def tokenizeForInference(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
index f4515c7530a6..5cc71d3d5cfb 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
@@ -107,6 +107,9 @@ case class CSVFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
SerializableConfiguration.broadcast(sparkSession.sparkContext,
hadoopConf)
val parsedOptions = getCsvOptions(sparkSession, options)
val isColumnPruningEnabled =
parsedOptions.isColumnPruningEnabled(requiredSchema)
+ // The effective `ignoredPathSegmentRegex` for skipping hidden archive
entries. Reused from the
+ // options' lazily-compiled Pattern (Serializable) instead of re-compiling
per file or entry.
+ val ignoredPathSegmentRegex = parsedOptions.ignoredPathSegmentRegexPattern
// Check a field requirement for corrupt records here to throw an
exception in a driver side
ExprUtils.verifyColumnNameOfCorruptRecord(dataSchema,
parsedOptions.columnNameOfCorruptRecord)
@@ -141,7 +144,7 @@ case class CSVFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
// archive reads are enabled; otherwise the file is parsed directly.
if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
CSVDataSource(parsedOptions).readArchive(
- conf, file, () => newParser(), getHeaderChecker, requiredSchema)
+ conf, file, () => newParser(), getHeaderChecker, requiredSchema,
ignoredPathSegmentRegex)
} else {
val parser = newParser()
val headerChecker = getHeaderChecker(file.start == 0, s"CSV file:
${file.urlEncodedPath}")
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
index bafc3a9ea52f..f87e155d41ab 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetUtils.scala
@@ -144,7 +144,10 @@ object ParquetUtils extends Logging {
val leaves = allFiles.toArray.sortBy(_.getPath.toString)
FileTypes(
- data = leaves.filterNot(f =>
isSummaryFile(f.getPath)).toImmutableArraySeq,
+ // Zero-length files (e.g. `_SUCCESS` markers surfaced by a relaxed
`ignoredPathSegmentRegex`)
+ // cannot contain a Parquet footer, so exclude them from schema
inference candidates.
+ data = leaves.filter(_.getLen > 0).filterNot(f =>
isSummaryFile(f.getPath))
+ .toImmutableArraySeq,
metadata =
leaves.filter(_.getPath.getName ==
ParquetFileWriter.PARQUET_METADATA_FILE)
.toImmutableArraySeq,
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
index 0af728c1958d..8941a4c8d8c7 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileTable.scala
@@ -23,6 +23,7 @@ import scala.jdk.CollectionConverters._
import org.apache.hadoop.fs.{FileStatus, Path}
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.FileSourceOptions
import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite,
Table, TableCapability}
import org.apache.spark.sql.connector.catalog.TableCapability._
import org.apache.spark.sql.connector.expressions.Transform
@@ -56,8 +57,11 @@ abstract class FileTable(
options.asScala.toMap, userSpecifiedSchema)
} else {
// This is a non-streaming file based datasource.
+ val ignoredPathSegmentRegex =
+ new FileSourceOptions(caseSensitiveMap).ignoredPathSegmentRegexPattern
val rootPathsSpecified = DataSource.checkAndGlobPathIfNecessary(paths,
hadoopConf,
- checkEmptyGlobPath = true, checkFilesExist = true, enableGlobbing =
globPaths)
+ checkEmptyGlobPath = true, checkFilesExist = true, enableGlobbing =
globPaths,
+ ignoredPathSegmentRegex = ignoredPathSegmentRegex)
val fileStatusCache = FileStatusCache.getOrCreate(sparkSession)
new InMemoryFileIndex(
sparkSession, rootPathsSpecified, caseSensitiveMap,
userSpecifiedSchema, fileStatusCache)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
index 22f3ae6126ee..34cc2088c955 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/FileBasedDataSourceSuite.scala
@@ -47,6 +47,7 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.TimestampNanosVal
+import org.apache.spark.util.HadoopFSUtils
class FileBasedDataSourceSuite extends SharedSparkSession
with AdaptiveSparkPlanHelper {
@@ -1429,6 +1430,169 @@ class FileBasedDataSourceSuite extends
SharedSparkSession
}
}
}
+
+ // Asserts the ignoredPathSegmentRegex contract for `format`: the default
regex hides the
+ // '_'-prefixed file; a never-matching per-read option or session conf each
surface it; the
+ // option overrides the conf.
+ private def checkIgnoredPathSegmentRegexContract(format: String, basePath:
String): Unit = {
+ val defaultRegex = HadoopFSUtils.DEFAULT_IGNORED_PATH_SEGMENT_REGEX
+
+ // Default: the hidden file is filtered out.
+ checkAnswer(spark.read.format(format).load(basePath), Seq(Row("visible")))
+
+ // A never-matching per-read option surfaces the hidden data.
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex",
"(?!)").format(format).load(basePath),
+ Seq(Row("visible"), Row("hidden")))
+
+ // A never-matching session conf surfaces the hidden data too.
+ withSQLConf(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key -> "(?!)") {
+ checkAnswer(
+ spark.read.format(format).load(basePath),
+ Seq(Row("visible"), Row("hidden")))
+ }
+
+ // Precedence: the per-read option overrides the session conf in both
directions.
+ withSQLConf(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key -> defaultRegex) {
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex",
"(?!)").format(format).load(basePath),
+ Seq(Row("visible"), Row("hidden")))
+ }
+ withSQLConf(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key -> "(?!)") {
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex",
defaultRegex).format(format).load(basePath),
+ Seq(Row("visible")))
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: surfaces hidden files only when
relaxed - json") {
+ withTempDir { dir =>
+ val basePath = dir.getCanonicalPath
+ // Write JSON files directly: a visible record and a '_'-prefixed hidden
record.
+ Files.write(new File(dir, "visible.json").toPath,
"""{"a":"visible"}""".getBytes)
+ Files.write(new File(dir, "_hidden.json").toPath,
"""{"a":"hidden"}""".getBytes)
+
+ checkIgnoredPathSegmentRegexContract("json", basePath)
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: surfaces hidden files only when
relaxed - parquet") {
+ withTempDir { dir =>
+ // Copy only the part-*.parquet from a scratch write into basePath --
this drops Spark's own
+ // _SUCCESS / _temporary marker files, so a relaxed
ignoredPathSegmentRegex surfaces only the
+ // intended hidden file. Flat layout (files directly under basePath)
mirrors the JSON test.
+ val basePath = dir.getCanonicalPath
+
+ def copyPartFile(value: String, destName: String): Unit = {
+ withTempDir { scratch =>
+ // withTempDir pre-creates the directory, so overwrite to avoid
PATH_ALREADY_EXISTS.
+ Seq(value).toDF("a").write.format("parquet").mode("overwrite")
+ .save(scratch.getCanonicalPath)
+ val partFile = scratch.listFiles()
+ .find(f => f.isFile && f.getName.startsWith("part-") &&
f.getName.endsWith(".parquet"))
+ .getOrElse(fail(s"no part file produced under
${scratch.getCanonicalPath}"))
+ Files.copy(partFile.toPath, new File(dir, destName).toPath)
+ }
+ }
+ copyPartFile("visible", "visible.parquet")
+ copyPartFile("hidden", "_hidden.parquet")
+
+ checkIgnoredPathSegmentRegexContract("parquet", basePath)
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: reads an unmodified Spark-written
parquet dir") {
+ withTempDir { dir =>
+ // A normal Spark write leaves a zero-length _SUCCESS marker next to the
part files. With
+ // a never-matching ignoredPathSegmentRegex, the marker is surfaced by
listing but must
+ // neither be picked as a schema inference footer candidate nor scanned
as data.
+ val basePath = dir.getCanonicalPath
+ // withTempDir pre-creates the directory, so overwrite to avoid
PATH_ALREADY_EXISTS.
+ Seq("a", "b").toDF("col").write.mode("overwrite").parquet(basePath)
+ assert(new File(dir, "_SUCCESS").exists())
+
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex", "(?!)").parquet(basePath),
+ Seq(Row("a"), Row("b")))
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: reads a partitioned Spark-written
parquet dir") {
+ withTempDir { dir =>
+ // partitionBy leaves the zero-length _SUCCESS marker at the table root
next to the 'p='
+ // partition dirs. Under a never-matching ignoredPathSegmentRegex the
marker survives listing,
+ // but partition discovery still works (parsePartition stops at the base
path) and the
+ // zero-length marker is neither a schema inference footer candidate nor
scanned as data.
+ val basePath = dir.getCanonicalPath
+ // withTempDir pre-creates the directory, so overwrite to avoid
PATH_ALREADY_EXISTS.
+ Seq((1, "a"), (2, "b")).toDF("p",
"v").write.mode("overwrite").partitionBy("p")
+ .parquet(basePath)
+ assert(new File(dir, "_SUCCESS").exists())
+
+ // The read-back schema appends the discovered partition column after
the data column.
+ val expected = Seq(Row("a", 1), Row("b", 2))
+ checkAnswer(spark.read.parquet(basePath), expected)
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex", "(?!)").parquet(basePath),
+ expected)
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: custom regex replaces the default
filter - json") {
+ withTempDir { dir =>
+ val basePath = dir.getCanonicalPath
+ // The default '^[._]' hides both the '_'- and '.'-prefixed files. A
custom '^_' regex
+ // replaces it entirely: it keeps hiding the '_'-prefixed file but
surfaces the
+ // '.'-prefixed one.
+ Files.write(new File(dir, "visible.json").toPath,
"""{"a":"visible"}""".getBytes)
+ Files.write(new File(dir, "_under.json").toPath,
"""{"a":"under"}""".getBytes)
+ Files.write(new File(dir, ".dot.json").toPath,
"""{"a":"dot"}""".getBytes)
+
+ checkAnswer(spark.read.format("json").load(basePath),
Seq(Row("visible")))
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex",
"^_").format("json").load(basePath),
+ Seq(Row("visible"), Row("dot")))
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: invalid regexes are rejected") {
+ withTempDir { dir =>
+ val basePath = dir.getCanonicalPath
+ Files.write(new File(dir, "visible.json").toPath,
"""{"a":"visible"}""".getBytes)
+
+ // An invalid per-read regex is surfaced as a clear
IllegalArgumentException when the read
+ // resolves the relation, instead of a raw PatternSyntaxException deep
in file listing.
+ val optionError = intercept[IllegalArgumentException] {
+ spark.read.option("ignoredPathSegmentRegex",
"[").format("json").load(basePath)
+ }
+ assert(optionError.getMessage.contains("ignoredPathSegmentRegex"))
+
+ // The session conf rejects invalid values eagerly via checkValue.
+ val confError = intercept[IllegalArgumentException] {
+ spark.conf.set(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key, "[")
+ }
+ assert(confError.getMessage.contains("valid Java regular expression"))
+ }
+ }
+
+ test("Option ignoredPathSegmentRegex: an empty value disables hidden-file
filtering") {
+ withTempDir { dir =>
+ val basePath = dir.getCanonicalPath
+ Files.write(new File(dir, "visible.json").toPath,
"""{"a":"visible"}""".getBytes)
+ Files.write(new File(dir, "_hidden.json").toPath,
"""{"a":"hidden"}""".getBytes)
+
+ // The default '^[._]' hides the '_'-prefixed file.
+ checkAnswer(spark.read.format("json").load(basePath),
Seq(Row("visible")))
+ // An empty per-read option disables the generic hidden-file filter and
surfaces it.
+ checkAnswer(
+ spark.read.option("ignoredPathSegmentRegex",
"").format("json").load(basePath),
+ Seq(Row("visible"), Row("hidden")))
+ // An empty session conf does too.
+ withSQLConf(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key -> "") {
+ checkAnswer(spark.read.format("json").load(basePath),
Seq(Row("visible"), Row("hidden")))
+ }
+ }
+ }
}
object TestingUDT {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
index 792e5779d49b..2cbd1ddd9fd3 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReaderSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources
import java.io.{ByteArrayOutputStream, Closeable, File, FileOutputStream,
InputStream, OutputStream}
import java.nio.charset.StandardCharsets
import java.util.Properties
+import java.util.regex.Pattern
import java.util.zip.GZIPOutputStream
import scala.collection.mutable.ArrayBuffer
@@ -167,6 +168,21 @@ class ArchiveReaderSuite extends SparkFunSuite {
}
}
+ test("readEntries: a custom ignoredPathSegmentRegex pattern surfaces hidden
entries") {
+ withTempDir { dir =>
+ val tar = new File(dir, "custom-filter.tar")
+ writeTar(tar, Seq(textEntry("_SUCCESS", "marker"), textEntry("real.csv",
"kept")))
+ // "(?!)" matches nothing, so hidden-name filtering is disabled and only
the carve-outs in
+ // HadoopFSUtils.shouldFilterOutPathName still apply -- mirroring a
loose-file listing with
+ // the ignoredPathSegmentRegex option set to the same regex.
+ val entries = ArchiveReader(new Path(tar.toURI))
+ .readEntries(new Configuration(), Pattern.compile("(?!)")) { (name,
in) =>
+ Iterator.single((name, new String(readAll(in),
StandardCharsets.UTF_8)))
+ }.toList
+ assert(entries == Seq("_SUCCESS" -> "marker", "real.csv" -> "kept"))
+ }
+ }
+
test("readEntries: advances lazily, one entry at a time") {
withTempDir { dir =>
val tar = new File(dir, "lazy.tar")
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
index f4de8a52810e..6f72c31c6d4d 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileIndexSuite.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.functions.col
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{IntegerType, StringType, StructField,
StructType}
-import org.apache.spark.util.KnownSizeEstimation
+import org.apache.spark.util.{HadoopFSUtils, KnownSizeEstimation}
class FileIndexSuite extends SharedSparkSession {
@@ -646,9 +646,10 @@ class FileIndexSuite extends SharedSparkSession {
}
test("SPARK-40667: validate FileIndex Options") {
- assert(FileIndexOptions.getAllOptions.size == 8)
+ assert(FileIndexOptions.getAllOptions.size == 9)
// Please add validation on any new FileIndex options here
assert(FileIndexOptions.isValidOption("ignoreMissingFiles"))
+ assert(FileIndexOptions.isValidOption("ignoredPathSegmentRegex"))
assert(FileIndexOptions.isValidOption("ignoreInvalidPartitionPaths"))
assert(FileIndexOptions.isValidOption("timeZone"))
assert(FileIndexOptions.isValidOption("recursiveFileLookup"))
@@ -703,6 +704,180 @@ class FileIndexSuite extends SharedSparkSession {
assert(fileIndex2a != fileIndex2b)
}
}
+
+ test("InMemoryFileIndex: ignoredPathSegmentRegex option surfaces hidden
files in allFiles()") {
+ withTempDir { dir =>
+ stringToFile(new File(dir, "data.txt"), "data")
+ stringToFile(new File(dir, "_hidden"), "hidden")
+ stringToFile(new File(dir, ".dot"), "dot")
+ stringToFile(new File(dir, "x._COPYING_"), "copying")
+ val path = new Path(dir.getCanonicalPath)
+
+ // Default: hidden files are filtered out, only the regular file is
listed.
+ val defaultIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty,
None)
+ assert(defaultIndex.allFiles().map(_.getPath.getName).toSet ===
Set("data.txt"))
+
+ // With a never-matching regex, the hidden file and dot file surface,
but the copying
+ // file stays hidden: the '._COPYING_' carve-out is not overridable.
+ val hiddenIndex = new InMemoryFileIndex(
+ spark, Seq(path), Map("ignoredPathSegmentRegex" -> "(?!)"), None)
+ assert(hiddenIndex.allFiles().map(_.getPath.getName).toSet ===
+ Set("data.txt", "_hidden", ".dot"))
+
assert(!hiddenIndex.allFiles().map(_.getPath.getName).contains("x._COPYING_"))
+ }
+ }
+
+ test("InMemoryFileIndex: ignoredPathSegmentRegex option controls partition
dirs with hidden " +
+ "data files") {
+ withTempDir { dir =>
+ // 'part=1' holds a single hidden ('_'-prefixed) file: filtered out by
default (so no
+ // partition is discovered), but surfaced and counted as data with a
never-matching regex.
+ val partitionDir = new File(dir, "part=1")
+ assert(partitionDir.mkdir())
+ stringToFile(new File(partitionDir, "_data.txt"), "data")
+ val path = new Path(dir.getCanonicalPath)
+
+ // Default: the hidden data file is filtered out, so no files /
partitions.
+ val defaultIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty,
None)
+ assert(defaultIndex.allFiles().isEmpty)
+ assert(defaultIndex.partitionSpec().partitions.isEmpty)
+
+ // With the option set, '_data.txt' is a data file (covers isDataPath)
and 'part=1' is a
+ // partition directory.
+ val hiddenIndex = new InMemoryFileIndex(
+ spark, Seq(path), Map("ignoredPathSegmentRegex" -> "(?!)"), None)
+ assert(hiddenIndex.allFiles().map(_.getPath.getName).toSet ===
Set("_data.txt"))
+ val partitionValues =
hiddenIndex.partitionSpec().partitions.map(_.values)
+ assert(partitionValues.length == 1)
+ assert(partitionValues(0).numFields == 1)
+ assert(partitionValues(0).getInt(0) == 1)
+ assert(hiddenIndex.partitionSpec().partitionColumns.fieldNames.toSeq ===
Seq("part"))
+ }
+ }
+
+ test("InMemoryFileIndex: ignoredPathSegmentRegex with a hidden dir next to
partition dirs") {
+ withTempDir { dir =>
+ // A hidden directory holding files next to 'part=1' becomes a leaf data
dir when the
+ // regex never matches, so partition discovery sees conflicting base
paths.
+ val partitionDir = new File(dir, "part=1")
+ assert(partitionDir.mkdir())
+ stringToFile(new File(partitionDir, "data.txt"), "data")
+ val strayDir = new File(dir, "_stray")
+ assert(strayDir.mkdir())
+ stringToFile(new File(strayDir, "extra.txt"), "extra")
+ val path = new Path(dir.getCanonicalPath)
+
+ // Default: the hidden directory is filtered out and only the partition
data is listed.
+ val defaultIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty,
None)
+ assert(defaultIndex.allFiles().map(_.getPath.getName).toSet ===
Set("data.txt"))
+ assert(defaultIndex.partitionSpec().partitionColumns.fieldNames.toSeq
=== Seq("part"))
+
+ // Never-matching regex: '_stray' is treated as a leaf data dir and
discovered as its own
+ // base path, so partition discovery fails with conflicting directory
structures.
+ val hiddenIndex = new InMemoryFileIndex(
+ spark, Seq(path), Map("ignoredPathSegmentRegex" -> "(?!)"), None)
+ val ex = intercept[SparkRuntimeException] {
+ hiddenIndex.partitionSpec()
+ }
+ assert(ex.getMessage.contains("Conflicting directory structures
detected"))
+
+ // Never-matching regex + ignoreInvalidPartitionPaths=true: the invalid
'_stray' base path
+ // is ignored, 'part' is still discovered, and both files surface.
+ val recoveredIndex = new InMemoryFileIndex(
+ spark,
+ Seq(path),
+ Map("ignoredPathSegmentRegex" -> "(?!)", "ignoreInvalidPartitionPaths"
-> "true"),
+ None)
+ assert(recoveredIndex.partitionSpec().partitionColumns.fieldNames.toSeq
=== Seq("part"))
+ assert(recoveredIndex.allFiles().map(_.getPath.getName).toSet ===
+ Set("data.txt", "extra.txt"))
+ }
+ }
+
+ test("InMemoryFileIndex: ignoredPathSegmentRegex with a _SUCCESS marker next
to partition dirs") {
+ withTempDir { dir =>
+ // Unlike the hidden directory above, a hidden file at the table root
surfaces in the
+ // listing under a never-matching regex but does not break partition
discovery:
+ // parsePartition stops at the base path, so 'part' is still inferred.
+ val partitionDir = new File(dir, "part=1")
+ assert(partitionDir.mkdir())
+ stringToFile(new File(partitionDir, "data.txt"), "data")
+ assert(new File(dir, "_SUCCESS").createNewFile())
+ val path = new Path(dir.getCanonicalPath)
+
+ val hiddenIndex = new InMemoryFileIndex(
+ spark, Seq(path), Map("ignoredPathSegmentRegex" -> "(?!)"), None)
+ assert(hiddenIndex.allFiles().map(_.getPath.getName).toSet ===
Set("data.txt", "_SUCCESS"))
+ assert(hiddenIndex.partitionSpec().partitionColumns.fieldNames.toSeq ===
Seq("part"))
+ }
+ }
+
+ test("InMemoryFileIndex: ignoredPathSegmentRegex option overrides the SQL
conf") {
+ withTempDir { dir =>
+ stringToFile(new File(dir, "data.txt"), "data")
+ stringToFile(new File(dir, "_hidden"), "hidden")
+ val path = new Path(dir.getCanonicalPath)
+
+ val defaultRegex = HadoopFSUtils.DEFAULT_IGNORED_PATH_SEGMENT_REGEX
+
+ def hiddenListed(sqlConf: String, options: Map[String, String]): Boolean
= {
+ withSQLConf(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key -> sqlConf) {
+ val index = new InMemoryFileIndex(spark, Seq(path), options, None)
+ index.allFiles().map(_.getPath.getName).contains("_hidden")
+ }
+ }
+
+ // conf=default, option=never-matching -> option wins, hidden file
listed.
+ assert(hiddenListed(defaultRegex, Map("ignoredPathSegmentRegex" ->
"(?!)")))
+ // conf=never-matching, option=default -> option wins, hidden file not
listed.
+ assert(!hiddenListed("(?!)", Map("ignoredPathSegmentRegex" ->
defaultRegex)))
+ // conf=never-matching, no option -> conf takes effect, hidden file
listed.
+ assert(hiddenListed("(?!)", Map.empty))
+ // conf=default, no option -> default, hidden file not listed.
+ assert(!hiddenListed(defaultRegex, Map.empty))
+ }
+ }
+
+ test("InMemoryFileIndex: non-default ignoredPathSegmentRegex bypasses the
FileStatusCache") {
+ withTempDir { dir =>
+ stringToFile(new File(dir, "data.txt"), "data")
+ stringToFile(new File(dir, "_hidden.txt"), "hidden")
+ val path = new Path(dir.getCanonicalPath)
+ val fileStatusCache = FileStatusCache.getOrCreate(spark)
+
+ // Populate the cache with the default-filtered listing.
+ val defaultIndex = new InMemoryFileIndex(spark, Seq(path), Map.empty,
None, fileStatusCache)
+ assert(defaultIndex.allFiles().map(_.getPath.getName).toSet ===
Set("data.txt"))
+
+ // A non-default filter over the same path and cache must not serve the
cached
+ // default-filtered listing: the hidden file surfaces.
+ val hiddenIndex = new InMemoryFileIndex(
+ spark, Seq(path), Map("ignoredPathSegmentRegex" -> "(?!)"), None,
fileStatusCache)
+ assert(hiddenIndex.allFiles().map(_.getPath.getName).toSet ===
+ Set("data.txt", "_hidden.txt"))
+ }
+ }
+
+ test("InMemoryFileIndex: metadata summary files are never data paths
regardless of " +
+ "ignoredPathSegmentRegex") {
+ withTempDir { dir =>
+ stringToFile(new File(dir, "data.txt"), "data")
+ stringToFile(new File(dir, "_metadata"), "metadata")
+ stringToFile(new File(dir, ".dot"), "dot")
+ val path = new Path(dir.getCanonicalPath)
+
+ // Custom regex matching only dot files: '_metadata' still surfaces in
the listing (the
+ // shouldFilterOutPathName carve-out is not overridable), while '.dot'
stays hidden.
+ val index = new InMemoryFileIndex(
+ spark, Seq(path), Map("ignoredPathSegmentRegex" -> "^[.]"), None)
+ assert(index.allFiles().map(_.getPath.getName).toSet === Set("data.txt",
"_metadata"))
+
+ // Although listed, the summary file must never be scanned as data, even
when the regex
+ // does not classify it as hidden.
+ val dataFiles = index.listFiles(Nil,
Nil).flatMap(_.files).map(_.getPath.getName)
+ assert(dataFiles.toSet === Set("data.txt"))
+ }
+ }
}
object DeletionRaceFileSystem {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
index 4955a32c49b1..40a333407896 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala
@@ -2619,6 +2619,85 @@ class FileStreamSourceSuite extends FileStreamSourceTest
{
}
}
}
+
+ test("ignoredPathSegmentRegex option: hidden files are not streamed by
default") {
+ withTempDirs { case (src, tmp) =>
+ val textStream = createFileStream("text", src.getCanonicalPath)
+ val filtered = textStream.filter($"value" contains "keep")
+
+ // The '_'-prefixed file is a hidden file and must not be picked up by
the stream.
+ testStream(filtered)(
+ AddTextFileData("drop1\nkeep2", src, tmp, tmpFilePrefix = "_hidden"),
+ AddTextFileData("keep3", src, tmp),
+ CheckAnswer("keep3")
+ )
+ }
+ }
+
+ test("ignoredPathSegmentRegex option: hidden files are streamed with a
never-matching regex") {
+ withTempDirs { case (src, tmp) =>
+ val textStream = createFileStream(
+ "text", src.getCanonicalPath, options = Map("ignoredPathSegmentRegex"
-> "(?!)"))
+ val filtered = textStream.filter($"value" contains "keep")
+
+ // With a never-matching regex, the '_'-prefixed file participates in
streaming.
+ testStream(filtered)(
+ AddTextFileData("drop1\nkeep2", src, tmp, tmpFilePrefix = "_hidden"),
+ CheckAnswer("keep2"),
+ AddTextFileData("keep3", src, tmp),
+ CheckAnswer("keep2", "keep3")
+ )
+ }
+ }
+
+ test("ignoredPathSegmentRegex SQL conf: hidden files are streamed with a
never-matching regex") {
+ withSQLConf(SQLConf.IGNORED_PATH_SEGMENT_REGEX.key -> "(?!)") {
+ withTempDirs { case (src, tmp) =>
+ val textStream = createFileStream("text", src.getCanonicalPath)
+ val filtered = textStream.filter($"value" contains "keep")
+
+ // With the session conf set, the '_'-prefixed file participates in
streaming even
+ // though the ignoredPathSegmentRegex option is not set.
+ testStream(filtered)(
+ AddTextFileData("drop1\nkeep2", src, tmp, tmpFilePrefix = "_hidden"),
+ CheckAnswer("keep2"),
+ AddTextFileData("keep3", src, tmp),
+ CheckAnswer("keep2", "keep3")
+ )
+ }
+ }
+ }
+
+ test("ignoredPathSegmentRegex option does not surface another query's
_spark_metadata as data") {
+ // Regression: ignoredPathSegmentRegex must not change _spark_metadata
handling -- a reader on a
+ // sink's output dir still consults the metadata log rather than surfacing
the log files as
+ // data.
+ withSQLConf(SQLConf.FILESTREAM_SINK_METADATA_IGNORED.key -> "false") {
+ withTempDirs { case (outputDir, checkpointDir) =>
+ val source = MemoryStream[String]
+ val writer = source
+ .toDF()
+ .writeStream
+ .option("checkpointLocation", checkpointDir.getCanonicalPath)
+ .format("parquet")
+ .start(outputDir.getCanonicalPath)
+ try {
+ source.addData("keep1", "keep2")
+ writer.processAllAvailable()
+ } finally {
+ writer.stop()
+ }
+
+ // The sink must have created a _spark_metadata directory.
+ assert(new File(outputDir, FileStreamSink.metadataDir).exists())
+
+ // A never-matching regex returns the sink's data, not the
_spark_metadata log files.
+ val df = spark.read.option("ignoredPathSegmentRegex", "(?!)")
+ .parquet(outputDir.getCanonicalPath)
+ checkAnswer(df, Seq(Row("keep1"), Row("keep2")))
+ }
+ }
+ }
}
@SlowSQLTest
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
