You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/10 11:35:22 UTC
(spark) branch branch-4.3 updated: [SPARK-58609][SQL] Add archivePathFilter option to select inner archive entries
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new fc8f161f004b [SPARK-58609][SQL] Add archivePathFilter option to select inner archive entries
fc8f161f004b is described below
commit fc8f161f004bcc70f3ad20021d93eace5356be6b
Author: akshatshenoi-db <ak...@databricks.com>
AuthorDate: Mon Aug 10 19:34:33 2026 +0800
[SPARK-58609][SQL] Add archivePathFilter option to select inner archive entries
### What changes were proposed in this pull request?
This PR adds a new file-source option `archivePathFilter`: a glob that selects which **inner entries** of an archive are ingested, matched against each entry's full path within the archive, so patterns like `subdir/*` and `*/*.csv` work.
It continues the archive-read series (SPARK-57135 / SPARK-57321 CSV, SPARK-57419 JSON, SPARK-57478 text, SPARK-57479 XML, SPARK-57481 Avro, SPARK-57590 Parquet, SPARK-57591 ORC, SPARK-58382 binaryFile, SPARK-58110 trait extraction).
Details:
- `FileSourceOptions` gains the `archivePathFilter` option (the glob string) plus `archivePathFilterPattern`, a `transient lazy val` holding the compiled Hadoop `GlobPattern`. That caches per `FileSourceOptions` instance, so the reads sharing one options object reuse a single matcher instead of recompiling per archive; it is `transient` because `GlobPattern` is not serializable, so the string travels and an executor recompiles on first use. Two kinds of call site cannot reach that val [...]
- `SupportsArchiveFormat.readArchiveEntries` and the random-access `readLocalizedEntries` / `localizeEntries` take an optional filter (default `None`, so existing callers are unaffected), applied in `shouldSkipEntry`.
- The option is threaded from the read paths of CSV, JSON, text, XML, Avro, binaryFile, Parquet and ORC.
The filter is applied **in addition to** `ignoredPathSegmentRegex` (both must pass), so hidden entries stay hidden even when they match the glob.
### Why are the changes needed?
`pathGlobFilter` cannot express this. It is applied during file listing, so it must already match the archive file itself for that archive to be read at all; it cannot simultaneously select a subset of the entries *inside* the archive. Without `archivePathFilter` there is no way to read only part of an archive, so a user wanting a few entries has to read and discard the rest.
### Does this PR introduce _any_ user-facing change?
Yes. A new read option `archivePathFilter` (a glob string, no default) is available to the file sources that support archive reads. When set, only archive entries whose full inner path matches the glob are ingested. Reads that do not set it are unaffected, and the whole archive-read feature remains gated by `spark.sql.files.archive.reader.enabled` (default `false`).
### How was this patch tested?
New shared cases in `ArchiveReadSuiteBase`, which run for every (format, container) pair -- CSV, JSON, XML, text, Avro, Parquet, ORC and binaryFile across tar, zip and 7z:
- selection by inner path (`sub/*`),
- an extension glob matching across subdirectories, including an entry that does not match the glob,
- a glob matching no entry yields zero rows,
- composition with `ignoredPathSegmentRegex` (a hidden entry matching the glob is still skipped).
New unit cases in `SupportsArchiveFormatSuite` cover the entry-skip logic directly (glob matching the full path, `*` crossing `/`, no-match, interaction with hidden-entry filtering) and invalid-glob rejection.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57808 from akshatshenoi-db/archive-pathfilter.
Lead-authored-by: akshatshenoi-db <ak...@databricks.com>
Co-authored-by: Wenchen Fan <cl...@gmail.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit 7d5a45413d8bc87df82ece22a728af020133238f)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/FileSourceOptions.scala | 42 +++++++++-
.../org/apache/spark/sql/avro/AvroFileFormat.scala | 4 +-
.../org/apache/spark/sql/avro/AvroUtils.scala | 17 ++--
.../datasources/SupportsArchiveFormat.scala | 49 ++++++++----
.../datasources/binaryfile/BinaryFileFormat.scala | 7 +-
.../execution/datasources/csv/CSVDataSource.scala | 79 +++++++++++--------
.../execution/datasources/csv/CSVFileFormat.scala | 3 +-
.../datasources/json/JsonDataSource.scala | 41 ++++++----
.../datasources/json/JsonFileFormat.scala | 3 +-
.../execution/datasources/orc/OrcFileFormat.scala | 7 +-
.../sql/execution/datasources/orc/OrcUtils.scala | 19 ++++-
.../datasources/parquet/ParquetFileFormat.scala | 30 ++++++--
.../datasources/text/TextFileFormat.scala | 4 +-
.../execution/datasources/xml/XmlDataSource.scala | 43 +++++++----
.../execution/datasources/xml/XmlFileFormat.scala | 3 +-
.../datasources/ArchiveReadSuiteBase.scala | 65 ++++++++++++++++
.../datasources/BinaryFileArchiveReadBase.scala | 13 ++++
.../datasources/SupportsArchiveFormatSuite.scala | 90 +++++++++++++++++++---
18 files changed, 408 insertions(+), 111 deletions(-)
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 62db2d90ec8f..1040bb031287 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
@@ -18,7 +18,11 @@ package org.apache.spark.sql.catalyst
import java.util.regex.{Pattern, PatternSyntaxException}
-import org.apache.spark.sql.catalyst.FileSourceOptions.{IGNORE_CORRUPT_FILES, IGNORE_MISSING_FILES, IGNORED_PATH_SEGMENT_REGEX}
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.fs.GlobPattern
+
+import org.apache.spark.sql.catalyst.FileSourceOptions.{ARCHIVE_PATH_FILTER, 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}
@@ -66,12 +70,37 @@ class FileSourceOptions(
*/
lazy val ignoredPathSegmentRegexPattern: Pattern =
FileSourceOptions.compileIgnoredPathSegmentRegex(ignoredPathSegmentRegex)
+
+ /**
+ * Glob selecting which inner archive entries to read, matched against each entry's full path
+ * within the archive (e.g. `subdir/*`, `*/*.csv`). An empty value disables the filter, matching
+ * how an empty [[ignoredPathSegmentRegex]] is treated. Validated here so an invalid glob fails on
+ * the driver.
+ */
+ val archivePathFilter: Option[String] = {
+ val glob = parameters.get(ARCHIVE_PATH_FILTER).filter(_.nonEmpty)
+ glob.foreach(FileSourceOptions.compileArchivePathFilter)
+ glob
+ }
+
+ /**
+ * The effective [[archivePathFilter]] is compiled once per instance of this class, so the archive
+ * reads sharing one options object reuse a single matcher rather than re-compiling per archive.
+ * `transient` because Hadoop's `GlobPattern` is not serializable, so an executor recompiles from
+ * [[archivePathFilter]] on first use. Paths that cannot reach this value -- the schema-inference
+ * RDDs, which would have to capture the matcher in a closure, and the parallel footer/schema
+ * readers, whose signatures are fixed -- carry [[archivePathFilter]] instead and compile it
+ * themselves.
+ */
+ @transient lazy val archivePathFilterPattern: Option[GlobPattern] =
+ archivePathFilter.map(FileSourceOptions.compileArchivePathFilter)
}
object FileSourceOptions {
val IGNORE_CORRUPT_FILES = "ignoreCorruptFiles"
val IGNORE_MISSING_FILES = "ignoreMissingFiles"
val IGNORED_PATH_SEGMENT_REGEX = "ignoredPathSegmentRegex"
+ val ARCHIVE_PATH_FILTER = "archivePathFilter"
// A regex that never matches any name, used when the filter is disabled by an empty value.
private val DISABLED_FILTER_PATTERN = Pattern.compile("(?!)")
@@ -97,4 +126,15 @@ object FileSourceOptions {
}
}
}
+
+ /** Compiles `archivePathFilter` into a glob matcher, reporting an invalid glob clearly. */
+ def compileArchivePathFilter(glob: String): GlobPattern = {
+ try {
+ new GlobPattern(glob)
+ } catch {
+ case NonFatal(e) =>
+ throw new IllegalArgumentException(
+ s"The '$ARCHIVE_PATH_FILTER' value '$glob' is not a valid glob.", e)
+ }
+ }
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
index c2a05a8c9d8b..598872407096 100755
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
@@ -199,7 +199,9 @@ private[sql] class AvroFileFormat extends FileFormat
} else {
new NoopFilters
}
- SupportsArchiveFormat.readArchiveEntries(file.toPath, conf) { (_, in) =>
+ val entryGlob = parsedOptions.archivePathFilterPattern
+ SupportsArchiveFormat.readArchiveEntries(
+ file.toPath, conf, archivePathFilter = entryGlob) { (_, in) =>
val datumReader = userProvidedSchema match {
case Some(schema) => new GenericDatumReader[GenericRecord](schema)
case None => new GenericDatumReader[GenericRecord]()
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index a1bf478f14f1..266e6ee835ce 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -27,7 +27,7 @@ import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.mapred.{AvroOutputFormat, FsInput}
import org.apache.avro.mapreduce.AvroJob
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.fs.{FileStatus, GlobPattern, Path}
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkException, SparkIllegalArgumentException}
@@ -97,7 +97,8 @@ private[sql] object AvroUtils extends Logging {
}
if (archives.nonEmpty) {
inferAvroSchemaFromArchives(archives, nonArchives, conf, parsedOptions.ignoreExtension,
- fileSourceOptions.ignoreCorruptFiles, fileSourceOptions.ignoreMissingFiles)
+ fileSourceOptions.ignoreCorruptFiles, fileSourceOptions.ignoreMissingFiles,
+ fileSourceOptions.archivePathFilterPattern)
} else {
inferAvroSchemaFromFiles(files, conf, parsedOptions.ignoreExtension,
fileSourceOptions.ignoreCorruptFiles)
@@ -250,10 +251,12 @@ private[sql] object AvroUtils extends Logging {
conf: Configuration,
ignoreExtension: Boolean,
ignoreCorruptFiles: Boolean,
- ignoreMissingFiles: Boolean): Schema = {
+ ignoreMissingFiles: Boolean,
+ archivePathFilter: Option[GlobPattern]): Schema = {
archives.iterator
.flatMap { f =>
- firstArchiveEntrySchema(f.getPath, conf, ignoreCorruptFiles, ignoreMissingFiles)
+ firstArchiveEntrySchema(
+ f.getPath, conf, ignoreCorruptFiles, ignoreMissingFiles, archivePathFilter)
}
.nextOption()
.getOrElse {
@@ -274,11 +277,13 @@ private[sql] object AvroUtils extends Logging {
path: Path,
conf: Configuration,
ignoreCorruptFiles: Boolean,
- ignoreMissingFiles: Boolean): Option[Schema] = {
+ ignoreMissingFiles: Boolean,
+ archivePathFilter: Option[GlobPattern]): Option[Schema] = {
try {
// `readArchiveEntries` returns a Closeable iterator; take the first entry's schema and close
// it so the archive stream is released without draining the remaining entries.
- val entries = SupportsArchiveFormat.readArchiveEntries(path, conf) { (_, in) =>
+ val entries = SupportsArchiveFormat.readArchiveEntries(
+ path, conf, archivePathFilter = archivePathFilter) { (_, in) =>
val stream = new DataFileStream[GenericRecord](in, new GenericDatumReader[GenericRecord]())
try {
Iterator.single(stream.getSchema)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
index ff25f957c29d..dbfcff3f1fa5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala
@@ -34,7 +34,7 @@ import org.apache.commons.compress.archivers.zip.ZipFile
import org.apache.commons.io.ByteOrderMark
import org.apache.commons.io.input.{BOMInputStream, CloseShieldInputStream}
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FSDataInputStream, Path}
+import org.apache.hadoop.fs.{FSDataInputStream, GlobPattern, Path}
import org.apache.hadoop.io.Text
import org.apache.hadoop.util.LineReader
@@ -104,16 +104,18 @@ trait SupportsArchiveFormat extends Logging {
* Reads an archive by unpacking each entry to a temp file and applying `readEntry`, for a
* format that needs a complete file on disk (random access).
*
- * @param file the archive as a [[PartitionedFile]]
- * @param conf Hadoop configuration used to open the archive
- * @param tempPrefix prefix for the per-task temp dir the entries are unpacked into
- * @param readEntry reads one unpacked entry file into rows
+ * @param file the archive as a [[PartitionedFile]]
+ * @param conf Hadoop configuration used to open the archive
+ * @param tempPrefix prefix for the per-task temp dir the entries are unpacked into
+ * @param archivePathFilter optional glob matched against the entry's full path
+ * @param readEntry reads one unpacked entry file into rows
* @return iterator of rows across all entries
*/
protected def readLocalizedEntries(
file: PartitionedFile,
conf: Configuration,
- tempPrefix: String)(
+ tempPrefix: String,
+ archivePathFilter: Option[GlobPattern])(
readEntry: PartitionedFile => Iterator[InternalRow]): Iterator[InternalRow] = {
val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), tempPrefix)
// Register cleanup before constructing `entries`, which can throw before returning an iterator
@@ -122,7 +124,8 @@ trait SupportsArchiveFormat extends Logging {
Utils.deleteRecursively(tempDir)
})
val entries =
- try SupportsArchiveFormat.localizeEntries(file.toPath, conf, tempDir, archiveEntryFilter)
+ try SupportsArchiveFormat.localizeEntries(
+ file.toPath, conf, tempDir, archiveEntryFilter, archivePathFilter)
catch {
case NonFatal(e) =>
Utils.deleteRecursively(tempDir)
@@ -342,10 +345,16 @@ object SupportsArchiveFormat {
*
* @param entry the archive entry to test
* @param ignoredPathSegmentRegex per-segment filter matched against each `/`-separated component
- * @return true if the entry is a directory or any path component is filtered out
+ * @param archivePathFilter optional glob matched against the entry's full path
+ * @return true if the entry is a directory, any path component is filtered out, or the entry's
+ * path does not match `archivePathFilter`
*/
- private def shouldSkipEntry(entry: ArchiveEntry, ignoredPathSegmentRegex: Pattern): Boolean = {
+ private def shouldSkipEntry(
+ entry: ArchiveEntry,
+ ignoredPathSegmentRegex: Pattern,
+ archivePathFilter: Option[GlobPattern]): Boolean = {
if (entry.isDirectory) return true
+ if (archivePathFilter.exists(!_.matches(entry.getName))) return true
entry.getName.split("/").exists(c =>
c.nonEmpty && HadoopFSUtils.shouldFilterOutPathName(c, ignoredPathSegmentRegex))
}
@@ -359,13 +368,15 @@ object SupportsArchiveFormat {
* @param ignoredPathSegmentRegex per-segment filter for entries to skip (defaults to the
* `InMemoryFileIndex` filter); pass a custom one to match a
* loose-file scan
+ * @param archivePathFilter optional glob matched against the entry's full path
* @param parseEntry turns one entry's `(entry, stream)` into an iterator of results
* @return the concatenated results across kept entries, lazily one entry at a time
*/
def readArchiveEntries[T](
path: Path,
conf: Configuration,
- ignoredPathSegmentRegex: Pattern = HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern)(
+ ignoredPathSegmentRegex: Pattern = HadoopFSUtils.defaultIgnoredPathSegmentRegexPattern,
+ archivePathFilter: Option[GlobPattern])(
parseEntry: (ArchiveEntry, InputStream) => Iterator[T]): Iterator[T] = {
val archive = openArchiveStream(path, conf)
var closed = false
@@ -395,7 +406,9 @@ object SupportsArchiveFormat {
var next: (ArchiveEntry, InputStream) = null
while (next == null && archive.hasNext) {
val entry = archive.next()
- if (!shouldSkipEntry(entry._1, ignoredPathSegmentRegex)) next = entry
+ if (!shouldSkipEntry(entry._1, ignoredPathSegmentRegex, archivePathFilter)) {
+ next = entry
+ }
}
if (next == null) {
done = true
@@ -460,17 +473,19 @@ object SupportsArchiveFormat {
* companion so executor-side callers (a format's distributed archive inference) can use it
* without a trait instance.
*
- * @param path the archive path
- * @param conf Hadoop configuration used to open the archive
- * @param localDir directory the per-entry temp files are created under
- * @param entryFilter which entry names to keep
+ * @param path the archive path
+ * @param conf Hadoop configuration used to open the archive
+ * @param localDir directory the per-entry temp files are created under
+ * @param entryFilter which entry names to keep
+ * @param archivePathFilter optional glob matched against the entry's full path
*/
def localizeEntries(
path: Path,
conf: Configuration,
localDir: File,
- entryFilter: String => Boolean): Iterator[(String, File)] =
- readArchiveEntries(path, conf) { (entry, in) =>
+ entryFilter: String => Boolean,
+ archivePathFilter: Option[GlobPattern]): Iterator[(String, File)] =
+ readArchiveEntries(path, conf, archivePathFilter = archivePathFilter) { (entry, in) =>
val name = entry.getName
if (entryFilter(name)) {
Iterator.single((name, copyEntryToLocalFile(in, localDir, name)))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala
index 33de63072ef3..c0e50ee02e52 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/binaryfile/BinaryFileFormat.scala
@@ -113,16 +113,17 @@ case class BinaryFileFormat() extends FileFormat
val caseInsensitiveOptions = CaseInsensitiveMap(options)
val archiveReadEnabled = !caseInsensitiveOptions.get(WHOLE_FILE).forall(_.toBoolean) &&
getSqlConf(sparkSession).getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+ val fileSourceOptions = new FileSourceOptions(caseInsensitiveOptions)
file: PartitionedFile => {
val path = file.toPath
val fs = path.getFileSystem(broadcastedHadoopConf.value.value)
val status = fs.getFileStatus(path)
if (archiveReadEnabled && SupportsArchiveFormat.isArchivePath(path)) {
- val ignoredPathSegmentRegex =
- new FileSourceOptions(caseInsensitiveOptions).ignoredPathSegmentRegexPattern
SupportsArchiveFormat.readArchiveEntries(
- path, broadcastedHadoopConf.value.value, ignoredPathSegmentRegex) { (entry, in) =>
+ path, broadcastedHadoopConf.value.value,
+ fileSourceOptions.ignoredPathSegmentRegexPattern,
+ fileSourceOptions.archivePathFilterPattern) { (entry, in) =>
val entryStatus = new FileStatus(
entry.getSize, false, 0, 0, status.getModificationTime,
new Path(s"${status.getPath}!/${entry.getName}"))
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 ddaf8ba2539c..d2075a3304df 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
@@ -25,7 +25,7 @@ import scala.util.control.NonFatal
import com.univocity.parsers.csv.CsvParser
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.fs.{FileStatus, GlobPattern, Path}
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
@@ -36,7 +36,7 @@ import org.apache.spark.internal.LogKeys.PATH
import org.apache.spark.paths.SparkPath
import org.apache.spark.rdd.{BinaryFileRDD, RDD}
import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.csv.{CSVHeaderChecker, CSVInferSchema, CSVOptions, UnivocityParser}
import org.apache.spark.sql.classic.ClassicConversions.castToImpl
import org.apache.spark.sql.errors.QueryExecutionErrors
@@ -118,6 +118,7 @@ abstract class CSVDataSource extends Serializable with Logging with SupportsArch
* @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.
+ * @param archivePathFilter optional glob matched against the entry's full path
*/
def readArchive(
conf: Configuration,
@@ -125,7 +126,8 @@ abstract class CSVDataSource extends Serializable with Logging with SupportsArch
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
requiredSchema: StructType,
- ignoredPathSegmentRegex: Pattern): Iterator[InternalRow]
+ ignoredPathSegmentRegex: Pattern,
+ archivePathFilter: Option[GlobPattern]): Iterator[InternalRow]
/**
* Shared driver used by the [[readArchive]] implementations: streams each non-skipped entry's
@@ -138,11 +140,12 @@ abstract class CSVDataSource extends Serializable with Logging with SupportsArch
file: PartitionedFile,
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
- ignoredPathSegmentRegex: Pattern)(
+ ignoredPathSegmentRegex: Pattern,
+ archivePathFilter: Option[GlobPattern])(
parseEntry: (UnivocityParser, CSVHeaderChecker, InputStream) => Iterator[InternalRow])
: Iterator[InternalRow] = {
SupportsArchiveFormat.readArchiveEntries(
- file.toPath, conf, ignoredPathSegmentRegex) { (entry, in) =>
+ file.toPath, conf, ignoredPathSegmentRegex, archivePathFilter) { (entry, in) =>
val headerChecker =
getHeaderChecker(true, s"CSV archive entry: ${file.urlEncodedPath}!/${entry.getName}")
val parser = getParser()
@@ -168,29 +171,41 @@ abstract class CSVDataSource extends Serializable with Logging with SupportsArch
inputPaths: Seq[FileStatus],
parsedOptions: CSVOptions): StructType = {
val baseRdd = CSVDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
- def tokens(dropHeader: Boolean): RDD[Array[String]] = baseRdd.flatMap { stream =>
- val path = new Path(stream.getPath())
- try {
- if (SupportsArchiveFormat.isArchivePath(path)) {
- SupportsArchiveFormat.readArchiveEntries(path, stream.getConfiguration) { (_, in) =>
- tokenizeForInference(in, dropHeader, parsedOptions)
+ // Inference must see the same entries the scan reads, so it honors archivePathFilter too.
+ // Capture the glob string: the compiled GlobPattern is not serializable, so each task
+ // compiles it once when the archive branch is taken.
+ val archivePathFilterGlob = parsedOptions.archivePathFilter
+ def tokens(dropHeader: Boolean): RDD[Array[String]] = baseRdd.mapPartitions { streams =>
+ // Compile at most once per partition: lazy so a partition of only loose files never
+ // compiles, while a partition with archives reuses one matcher across all of them.
+ lazy val archivePathFilter =
+ archivePathFilterGlob.map(FileSourceOptions.compileArchivePathFilter)
+ streams.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ try {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(
+ path, stream.getConfiguration, archivePathFilter = archivePathFilter) {
+ (_, in) =>
+ tokenizeForInference(in, dropHeader, parsedOptions)
+ }
+ } else {
+ tokenizeForInference(
+ CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path),
+ dropHeader, parsedOptions)
}
- } else {
- tokenizeForInference(
- CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path),
- dropHeader, parsedOptions)
+ } catch {
+ case e: FileNotFoundException if parsedOptions.ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH, stream.getPath())}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if parsedOptions.ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted input: ${MDC(PATH, stream.getPath())}", e)
+ Iterator.empty
+ case NonFatal(e) =>
+ throw QueryExecutionErrors.cannotReadFilesError(
+ e, SparkPath.fromPathString(stream.getPath()).urlEncoded)
}
- } catch {
- case e: FileNotFoundException if parsedOptions.ignoreMissingFiles =>
- logWarning(log"Skipped missing input: ${MDC(PATH, stream.getPath())}", e)
- Iterator.empty
- case e: FileNotFoundException => throw e
- case e @ (_: RuntimeException | _: IOException) if parsedOptions.ignoreCorruptFiles =>
- logWarning(log"Skipped the corrupted input: ${MDC(PATH, stream.getPath())}", e)
- Iterator.empty
- case NonFatal(e) =>
- throw QueryExecutionErrors.cannotReadFilesError(
- e, SparkPath.fromPathString(stream.getPath()).urlEncoded)
}
}
tokens(dropHeader = false).take(1).headOption match {
@@ -299,10 +314,12 @@ object TextInputCSVDataSource extends CSVDataSource {
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
requiredSchema: StructType,
- ignoredPathSegmentRegex: Pattern): Iterator[InternalRow] =
+ ignoredPathSegmentRegex: Pattern,
+ archivePathFilter: Option[GlobPattern]): 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, ignoredPathSegmentRegex) {
+ streamArchiveEntries(
+ conf, file, getParser, getHeaderChecker, ignoredPathSegmentRegex, archivePathFilter) {
(parser, headerChecker, in) =>
UnivocityParser.parseIterator(
entryLines(in, parser.options), parser, headerChecker, requiredSchema)
@@ -428,10 +445,12 @@ object MultiLineCSVDataSource extends CSVDataSource {
getParser: () => UnivocityParser,
getHeaderChecker: (Boolean, String) => CSVHeaderChecker,
requiredSchema: StructType,
- ignoredPathSegmentRegex: Pattern): Iterator[InternalRow] =
+ ignoredPathSegmentRegex: Pattern,
+ archivePathFilter: Option[GlobPattern]): 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, ignoredPathSegmentRegex) {
+ streamArchiveEntries(
+ conf, file, getParser, getHeaderChecker, ignoredPathSegmentRegex, archivePathFilter) {
(parser, headerChecker, in) =>
UnivocityParser.parseStream(in, parser, headerChecker, requiredSchema)
}
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 0bfee12d1666..388030dcf42d 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
@@ -142,7 +142,8 @@ case class CSVFileFormat() extends TextBasedFileFormat with DataSourceRegister {
// archive reads are enabled; otherwise the file is parsed directly.
if (parsedOptions.archiveFormatEnabled && SupportsArchiveFormat.isArchivePath(file.toPath)) {
CSVDataSource(parsedOptions).readArchive(
- conf, file, () => newParser(), getHeaderChecker, requiredSchema, ignoredPathSegmentRegex)
+ conf, file, () => newParser(), getHeaderChecker, requiredSchema, ignoredPathSegmentRegex,
+ parsedOptions.archivePathFilterPattern)
} 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/json/JsonDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
index 42568a2e6b0a..14e23a5fbdee 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonDataSource.scala
@@ -23,7 +23,7 @@ import scala.util.control.NonFatal
import com.fasterxml.jackson.core.{JsonFactory, JsonParser}
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.fs.{FileStatus, GlobPattern, Path}
import org.apache.hadoop.io.Text
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
@@ -35,7 +35,7 @@ import org.apache.spark.internal.LogKeys.PATH
import org.apache.spark.paths.SparkPath
import org.apache.spark.rdd.{BinaryFileRDD, RDD}
import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser, JacksonParser, JsonInferSchema, JSONOptions}
import org.apache.spark.sql.catalyst.util.FailureSafeParser
import org.apache.spark.sql.classic.ClassicConversions.castToImpl
@@ -84,13 +84,16 @@ abstract class JsonDataSource extends Serializable with Logging with SupportsArc
* and is intentionally left untouched.
*
* @param parser builds a fresh JSON parser for each entry.
+ * @param archivePathFilter optional glob matched against the entry's full path
*/
def readArchive(
conf: Configuration,
file: PartitionedFile,
parser: () => JacksonParser,
- schema: StructType): Iterator[InternalRow] =
- SupportsArchiveFormat.readArchiveEntries(file.toPath, conf) { (_, in) =>
+ schema: StructType,
+ archivePathFilter: Option[GlobPattern]): Iterator[InternalRow] =
+ SupportsArchiveFormat.readArchiveEntries(
+ file.toPath, conf, archivePathFilter = archivePathFilter) { (_, in) =>
readStream(in, parser(), schema)
}
@@ -284,22 +287,34 @@ object MultiLineJsonDataSource extends JsonDataSource {
inputPaths: Seq[FileStatus],
parsedOptions: JSONOptions): StructType = {
val baseRdd = JsonDataSource.createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ // Inference must see the same entries the scan reads, so it honors archivePathFilter too.
+ // Capture the glob string: the compiled GlobPattern is not serializable, so each task
+ // compiles it once when the archive branch is taken.
+ val archivePathFilterGlob = parsedOptions.archivePathFilter
val encoding = parsedOptions.encoding
val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
// An archive entry's stream is only valid until the shared cursor advances, so each document
// must be consumed before the next is pulled; `JsonInferSchema.infer` does.
- val docs: RDD[InputStream] = baseRdd.flatMap { stream =>
- val path = new Path(stream.getPath())
- skipInputOnError(stream.getPath(), ignoreMissingFiles, ignoreCorruptFiles) {
- if (SupportsArchiveFormat.isArchivePath(path)) {
- SupportsArchiveFormat.readArchiveEntries(path, stream.getConfiguration) { (_, in) =>
- Iterator.single(in)
+ val docs: RDD[InputStream] = baseRdd.mapPartitions { streams =>
+ // Compile at most once per partition: lazy so a partition of only loose files never
+ // compiles, while a partition with archives reuses one matcher across all of them.
+ lazy val archivePathFilter =
+ archivePathFilterGlob.map(FileSourceOptions.compileArchivePathFilter)
+ streams.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ skipInputOnError(stream.getPath(), ignoreMissingFiles, ignoreCorruptFiles) {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(
+ path, stream.getConfiguration, archivePathFilter = archivePathFilter) {
+ (_, in) =>
+ Iterator.single(in)
+ }
+ } else {
+ Iterator.single(
+ CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
}
- } else {
- Iterator.single(
- CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path))
}
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
index 4ede461a4d51..eeeb96d0e6e5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/json/JsonFileFormat.scala
@@ -109,7 +109,8 @@ case class JsonFileFormat() extends TextBasedFileFormat with DataSourceRegister
filters)
if (parsedOptions.archiveFormatEnabled && SupportsArchiveFormat.isArchivePath(file.toPath)) {
JsonDataSource(parsedOptions).readArchive(
- broadcastedHadoopConf.value.value, file, () => parser(), requiredSchema)
+ broadcastedHadoopConf.value.value, file, () => parser(), requiredSchema,
+ parsedOptions.archivePathFilterPattern)
} else {
JsonDataSource(parsedOptions).readFile(
broadcastedHadoopConf.value.value,
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
index 15d30aaab2a9..c069abe3b438 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
@@ -33,7 +33,7 @@ import org.apache.orc.mapreduce._
import org.apache.spark.TaskContext
import org.apache.spark.memory.MemoryMode
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
@@ -178,6 +178,7 @@ class OrcFileFormat
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
val orcFilterPushDown = sqlConf.orcFilterPushDown
val archiveFormatEnabled = sqlConf.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+ val fileSourceOptions = new FileSourceOptions(options)
def readSingleFile(file: PartitionedFile): Iterator[InternalRow] = {
val conf = broadcastedConf.value.value
@@ -257,7 +258,9 @@ class OrcFileFormat
(file: PartitionedFile) => {
if (archiveFormatEnabled && SupportsArchiveFormat.isArchivePath(file.toPath)) {
- readLocalizedEntries(file, broadcastedConf.value.value, "orc-archive") { entryFile =>
+ readLocalizedEntries(
+ file, broadcastedConf.value.value, "orc-archive",
+ fileSourceOptions.archivePathFilterPattern) { entryFile =>
readSingleFile(entryFile)
}
} else {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
index 69f930600ae9..cf5544f92301 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
@@ -27,7 +27,7 @@ import scala.util.control.NonFatal
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.fs.{FileStatus, GlobPattern, Path}
import org.apache.hadoop.hive.serde2.io.DateWritable
import org.apache.hadoop.io.{BooleanWritable, ByteWritable, DoubleWritable, FloatWritable, IntWritable, LongWritable, ShortWritable, WritableComparable}
import org.apache.hadoop.mapreduce.lib.input.FileSplit
@@ -188,9 +188,12 @@ object OrcUtils extends Logging {
val ignoreMissingFiles =
new FileSourceOptions(CaseInsensitiveMap(options)).ignoreMissingFiles
val archiveFormatEnabled = SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+ val archivePathFilter =
+ new FileSourceOptions(CaseInsensitiveMap(options)).archivePathFilterPattern
files.iterator.flatMap { file =>
if (archiveFormatEnabled && SupportsArchiveFormat.isArchivePath(file.getPath)) {
- readArchiveSchemas(conf, file, ignoreCorruptFiles, ignoreMissingFiles, stopAtFirst = true)
+ readArchiveSchemas(conf, file, ignoreCorruptFiles, ignoreMissingFiles,
+ stopAtFirst = true, archivePathFilter)
.headOption
} else {
readSchema(file.getPath, conf, ignoreCorruptFiles).map { schema =>
@@ -211,9 +214,15 @@ object OrcUtils extends Logging {
// Read outside `parmap`: its worker threads do not inherit the caller's `SQLConf` thread-local,
// so `SQLConf.get` there would fall back to defaults and never take the archive branch.
val archiveEnabled = SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+ // The signature is fixed by `SchemaMergeUtils.mergeSchemasInParallel`'s `schemaReader` type,
+ // so the glob comes from `conf`, which that caller builds with the user options via
+ // `newHadoopConfWithOptions`.
+ val archivePathFilter = Option(conf.get(FileSourceOptions.ARCHIVE_PATH_FILTER))
+ .filter(_.nonEmpty).map(FileSourceOptions.compileArchivePathFilter)
ThreadUtils.parmap(files, "readingOrcSchemas", 8) { currentFile =>
if (archiveEnabled && SupportsArchiveFormat.isArchivePath(currentFile.getPath)) {
readArchiveSchemas(conf, currentFile, ignoreCorruptFiles, ignoreMissingFiles,
+ archivePathFilter = archivePathFilter,
stopAtFirst = false)
} else {
OrcUtils.readSchema(currentFile.getPath, conf, ignoreCorruptFiles, ignoreMissingFiles)
@@ -233,13 +242,15 @@ object OrcUtils extends Logging {
archive: FileStatus,
ignoreCorruptFiles: Boolean,
ignoreMissingFiles: Boolean,
- stopAtFirst: Boolean): Seq[StructType] = {
+ stopAtFirst: Boolean,
+ archivePathFilter: Option[GlobPattern]): Seq[StructType] = {
val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), "orc-archive-infer")
// localizeEntries eagerly opens the first entry, so build it inside the try; the finally must
// still delete tempDir when a corrupt archive throws there.
var entries: Iterator[(String, File)] = Iterator.empty
try {
- entries = SupportsArchiveFormat.localizeEntries(archive.getPath, conf, tempDir, _ => true)
+ entries = SupportsArchiveFormat.localizeEntries(
+ archive.getPath, conf, tempDir, _ => true, archivePathFilter)
// With ignore flags off, a corrupt entry throws to the per-archive catch below. `.toList`
// reads every entry (whole archive atomic); `stopAtFirst` stays lazy and stops at the first.
val schemas = entries.flatMap { case (_, entryFile) =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index 235aa428fbce..a97cd7d57e41 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -27,7 +27,7 @@ import scala.util.{Failure, Try}
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.fs.{FileStatus, GlobPattern, Path}
import org.apache.hadoop.mapred.FileSplit
import org.apache.hadoop.mapreduce._
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
@@ -41,12 +41,12 @@ import org.apache.spark.{SparkEnv, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.{PATH, SCHEMA}
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.parser.LegacyTypeStringParser
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
-import org.apache.spark.sql.catalyst.util.{DateTimeUtils, RebaseDateTime}
+import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, DateTimeUtils, RebaseDateTime}
import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.parquet.types.ops.ParquetTypeOps
@@ -322,7 +322,9 @@ class ParquetFileFormat
// An archive is read by unpacking each Parquet entry to a local temp file and reading it with
// the plain reader (readSingleFile); readLocalizedEntries owns the unpack/iterate/cleanup.
def readArchiveFile(file: PartitionedFile): Iterator[InternalRow] =
- readLocalizedEntries(file, broadcastedHadoopConf.value.value, "parquet-archive") {
+ readLocalizedEntries(
+ file, broadcastedHadoopConf.value.value, "parquet-archive",
+ parquetOptions.archivePathFilterPattern) {
entryFile => readSingleFile(entryFile)
}
@@ -552,6 +554,11 @@ object ParquetFileFormat extends Logging {
ignoreCorruptFiles: Boolean,
ignoreMissingFiles: Boolean = false): Seq[Footer] = {
val archiveEnabled = SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+ // This signature is shared with `SchemaMergeUtils.mergeSchemasInParallel`'s `schemaReader`, so
+ // the glob comes from `conf`, which that caller builds with the user options via
+ // `newHadoopConfWithOptions`.
+ val archivePathFilter = Option(conf.get(FileSourceOptions.ARCHIVE_PATH_FILTER))
+ .filter(_.nonEmpty).map(FileSourceOptions.compileArchivePathFilter)
ThreadUtils.parmap(partFiles, "readingParquetFooters", 8,
preserveSparkThrowable = true) { currentFile =>
try {
@@ -561,7 +568,7 @@ object ParquetFileFormat extends Logging {
if (archiveEnabled && SupportsArchiveFormat.isArchivePath(currentFile.getPath)) {
// An archive is one file here; read each of its Parquet entries' footers (the archive is
// atomic under ignoreCorruptFiles, see readArchiveFooters).
- readArchiveFooters(conf, currentFile)
+ readArchiveFooters(conf, currentFile, archivePathFilter)
} else {
Seq(new Footer(currentFile.getPath,
ParquetFooterReader.readFooter(
@@ -591,13 +598,17 @@ object ParquetFileFormat extends Logging {
}
/** Reads every Parquet entry's footer in one archive. */
- private def readArchiveFooters(conf: Configuration, archive: FileStatus): Seq[Footer] = {
+ private def readArchiveFooters(
+ conf: Configuration,
+ archive: FileStatus,
+ archivePathFilter: Option[GlobPattern]): Seq[Footer] = {
val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), "parquet-archive-infer")
// localizeEntries eagerly opens/copies the first entry, so build it inside the try -- a corrupt
// archive throws there and the finally must still delete tempDir.
var entries: Iterator[(String, File)] = Iterator.empty
try {
- entries = SupportsArchiveFormat.localizeEntries(archive.getPath, conf, tempDir, _ => true)
+ entries = SupportsArchiveFormat.localizeEntries(
+ archive.getPath, conf, tempDir, _ => true, archivePathFilter)
entries.map { case (_, entryFile) =>
try {
val status = new FileStatus(entryFile.length(), false, 0, 0, entryFile.lastModified(),
@@ -654,6 +665,11 @@ object ParquetFileFormat extends Logging {
timestampNanosTypesEnabled = timestampNanosTypesEnabled,
respectUnknownTypeAnnotation = respectUnknownTypeAnnotation)
+ // readParquetFootersInParallel reads archivePathFilter from the conf (its signature is fixed
+ // by SchemaMergeUtils' schemaReader type), so put the option there.
+ new FileSourceOptions(CaseInsensitiveMap(parameters)).archivePathFilter.foreach(
+ conf.set(FileSourceOptions.ARCHIVE_PATH_FILTER, _))
+
readParquetFootersInParallel(conf, files, ignoreCorruptFiles, ignoreMissingFiles)
.map(ParquetFileFormat.readSchemaFromFooter(_, converter))
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
index c220dd5a957f..62d2cd4a2edd 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/text/TextFileFormat.scala
@@ -142,7 +142,9 @@ case class TextFileFormat()
textOptions: TextOptions): PartitionedFile => Iterator[UnsafeRow] = {
(file: PartitionedFile) => {
val confValue = conf.value.value
- SupportsArchiveFormat.readArchiveEntries(file.toPath, confValue) { (_, in) =>
+ val entryGlob = textOptions.archivePathFilterPattern
+ SupportsArchiveFormat.readArchiveEntries(
+ file.toPath, confValue, archivePathFilter = entryGlob) { (_, in) =>
// Each entry is read as a standalone text file, so it gets its own row writer, exactly as
// `readToUnsafeMem` builds one per file.
val emptyUnsafeRow = new UnsafeRow(0)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
index 14116279801d..4ec07656f3f4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlDataSource.scala
@@ -23,7 +23,7 @@ import java.nio.charset.{Charset, StandardCharsets}
import scala.util.control.NonFatal
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.hadoop.fs.{FileStatus, GlobPattern, Path}
import org.apache.hadoop.hdfs.BlockMissingException
import org.apache.hadoop.mapreduce.Job
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
@@ -34,7 +34,7 @@ import org.apache.spark.input.{PortableDataStream, StreamInputFormat}
import org.apache.spark.internal.Logging
import org.apache.spark.rdd.{BinaryFileRDD, RDD}
import org.apache.spark.sql.{Dataset, Encoders, SparkSession}
-import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.util.FailureSafeParser
import org.apache.spark.sql.catalyst.xml.{StaxXmlParser, StaxXMLRecordReader, XmlInferSchema, XmlOptions}
import org.apache.spark.sql.classic.ClassicConversions.castToImpl
@@ -81,13 +81,16 @@ abstract class XmlDataSource extends Serializable with Logging with SupportsArch
* `XmlFileFormat` read path supports archives; XML has no DSv2 reader.
*
* @param parser builds a fresh XML parser for each entry.
+ * @param archivePathFilter optional glob matched against the entry's full path
*/
def readArchive(
conf: Configuration,
file: PartitionedFile,
parser: () => StaxXmlParser,
- schema: StructType): Iterator[InternalRow] =
- SupportsArchiveFormat.readArchiveEntries(file.toPath, conf) { (_, in) =>
+ schema: StructType,
+ archivePathFilter: Option[GlobPattern]): Iterator[InternalRow] =
+ SupportsArchiveFormat.readArchiveEntries(
+ file.toPath, conf, archivePathFilter = archivePathFilter) { (_, in) =>
readStream(in, parser(), schema)
}
@@ -369,20 +372,32 @@ object MultiLineXmlDataSource extends XmlDataSource {
inputPaths: Seq[FileStatus],
parsedOptions: XmlOptions): StructType = {
val baseRdd = createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ // Inference must see the same entries the scan reads, so it honors archivePathFilter too.
+ // Capture the glob string: the compiled GlobPattern is not serializable, so each task
+ // compiles it once when the archive branch is taken.
+ val archivePathFilterGlob = parsedOptions.archivePathFilter
val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
- val tokenRDD: RDD[String] = baseRdd.flatMap { stream =>
- val path = new Path(stream.getPath())
- skipInputOnError(ignoreMissingFiles, ignoreCorruptFiles) {
- if (SupportsArchiveFormat.isArchivePath(path)) {
- SupportsArchiveFormat.readArchiveEntries(path, stream.getConfiguration) { (_, in) =>
- StaxXmlParser.tokenizeStream(in, parsedOptions)
+ val tokenRDD: RDD[String] = baseRdd.mapPartitions { streams =>
+ // Compile at most once per partition: lazy so a partition of only loose files never
+ // compiles, while a partition with archives reuses one matcher across all of them.
+ lazy val archivePathFilter =
+ archivePathFilterGlob.map(FileSourceOptions.compileArchivePathFilter)
+ streams.flatMap { stream =>
+ val path = new Path(stream.getPath())
+ skipInputOnError(ignoreMissingFiles, ignoreCorruptFiles) {
+ if (SupportsArchiveFormat.isArchivePath(path)) {
+ SupportsArchiveFormat.readArchiveEntries(
+ path, stream.getConfiguration, archivePathFilter = archivePathFilter) {
+ (_, in) =>
+ StaxXmlParser.tokenizeStream(in, parsedOptions)
+ }
+ } else {
+ StaxXmlParser.tokenizeStream(
+ CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path),
+ parsedOptions)
}
- } else {
- StaxXmlParser.tokenizeStream(
- CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path),
- parsedOptions)
}
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
index 26bf2cd3ccab..7823efd6ae81 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XmlFileFormat.scala
@@ -131,7 +131,8 @@ case class XmlFileFormat() extends TextBasedFileFormat with DataSourceRegister {
broadcastedHadoopConf.value.value,
file,
() => parser(),
- requiredSchema)
+ requiredSchema,
+ xmlOptions.archivePathFilterPattern)
} else {
XmlDataSource(xmlOptions).readFile(
broadcastedHadoopConf.value.value,
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
index d10476c52dea..01a2521a737b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
@@ -357,9 +357,74 @@ trait ArchiveReadSuiteBase extends QueryTest with SharedSparkSession {
}
}
+ // ----- shared archivePathFilter tests --------------------------------------
+
+ test("archivePathFilter selects inner entries by full path") {
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ s"top.$fileExtension" -> encodeFile(sampleDf((1, "top"))),
+ s"sub/keep.$fileExtension" -> encodeFile(sampleDf((2, "keep"))),
+ s"other/skip.$fileExtension" -> encodeFile(sampleDf((3, "skip")))))
+ // `sub/*` matches the full inner path, so only the entry under `sub/` is ingested.
+ checkAnswer(
+ read(archive.getCanonicalPath, Map("archivePathFilter" -> "sub/*")).select("id", "name"),
+ Seq(Row(2, "keep")))
+ }
+ }
+
+ test("archivePathFilter with an extension glob selects across subdirectories") {
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ s"top.$fileExtension" -> encodeFile(sampleDf((1, "top"))),
+ s"sub/nested.$fileExtension" -> encodeFile(sampleDf((2, "nested"))),
+ "sub/skip.other" -> encodeFile(sampleDf((3, "skip")))))
+ // `*` crosses `/`, so the glob keeps both entries of this extension at any depth, while the
+ // entry with a different extension is filtered out.
+ checkAnswer(
+ read(archive.getCanonicalPath, Map("archivePathFilter" -> s"*.$fileExtension"))
+ .select("id", "name"),
+ Seq(Row(1, "top"), Row(2, "nested")))
+ }
+ }
+
+ test("archivePathFilter matching no entry yields no rows") {
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(s"data.$fileExtension" -> encodeFile(sampleDf((1, "Alice")))))
+ checkAnswer(
+ read(archive.getCanonicalPath, Map("archivePathFilter" -> "nomatch/*")), Seq.empty[Row])
+ }
+ }
+
+ test("archivePathFilter applies in addition to ignoredPathSegmentRegex") {
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ s"keep/data.$fileExtension" -> encodeFile(sampleDf((1, "keep"))),
+ // Matches the glob, but the hidden-file filter still drops the `_`-prefixed entry.
+ s"keep/_hidden.$fileExtension" -> encodeFile(sampleDf((2, "hidden")))))
+ checkAnswer(
+ read(archive.getCanonicalPath, Map("archivePathFilter" -> "keep/*")).select("id", "name"),
+ Seq(Row(1, "keep")))
+ }
+ }
+
// ----- shared schema-inference tests (run when `supportsSchemaInference`) --
if (supportsSchemaInference) {
+ test("archivePathFilter applies to schema inference, not just the scan") {
+ // The excluded entry carries a column the kept entry lacks. Inference must skip it, otherwise
+ // the inferred schema is a superset of what the scan returns and `extra` reads back all-null.
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ s"keep/data.$fileExtension" -> encodeFile(sampleDf((1, "keep"))),
+ s"skip/data.$fileExtension" ->
+ encodeFile(Seq((2, "skip", "x")).toDF("id", "name", "extra"))))
+ val schema = inferredSchema(
+ Seq(archive.getCanonicalPath), Map("archivePathFilter" -> "keep/*"))
+ assert(!schema.fieldNames.contains("extra"),
+ s"inference read a filtered-out entry; got $schema")
+ }
+ }
+
test("archive infers the same schema as a directory of the same files") {
val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3, "Carol")))
.zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
index e23c61a8236a..e8fc1268abb8 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/BinaryFileArchiveReadBase.scala
@@ -157,6 +157,19 @@ trait BinaryFileArchiveReadBase extends QueryTest with SharedSparkSession {
}
}
+ test("wholeFile=false honors archivePathFilter, applied on top of hidden-entry filtering") {
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ "keep/a.bin" -> bytes("a"),
+ "keep/_hidden.bin" -> bytes("drop"), // matches the glob but hidden
+ "other/b.bin" -> bytes("drop")))
+ checkAnswer(
+ read(archive.getCanonicalPath, Map("wholeFile" -> "false", "archivePathFilter" -> "keep/*"))
+ .select("content"),
+ Seq(Row(bytes("a"))))
+ }
+ }
+
test("wholeFile=false enforces SOURCES_BINARY_FILE_MAX_LENGTH per entry") {
withArchiveFile() { archive =>
writeArchive(archive, Seq("big.bin" -> bytes("0123456789")))
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
index 0c362afd31d8..d9eae971e975 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
@@ -30,9 +30,10 @@ import org.apache.commons.compress.archivers.sevenz.{SevenZArchiveEntry, SevenZO
import org.apache.commons.compress.archivers.tar.{TarArchiveEntry, TarArchiveOutputStream}
import org.apache.commons.compress.archivers.zip.{ZipArchiveEntry, ZipArchiveOutputStream}
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.Path
+import org.apache.hadoop.fs.{GlobPattern, Path}
import org.apache.spark.{SparkFunSuite, SparkRuntimeException, TaskContext, TaskContextImpl}
+import org.apache.spark.sql.catalyst.FileSourceOptions
/**
* Unit tests for the streaming [[SupportsArchiveFormat]] engine: `isArchivePath` dispatch and
@@ -207,11 +208,20 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
/** Drains every entry into `(name, decodedText)` pairs through `SupportsArchiveFormat`. */
private def collect(file: File): Seq[(String, String)] =
- SupportsArchiveFormat.readArchiveEntries(new Path(file.toURI), new Configuration()) {
+ SupportsArchiveFormat.readArchiveEntries(
+ new Path(file.toURI), new Configuration(), archivePathFilter = None) {
(entry, in) =>
Iterator.single((entry.getName, new String(readAll(in), StandardCharsets.UTF_8)))
}.toList
+ /** Drains every entry through `SupportsArchiveFormat` under an `archivePathFilter` glob. */
+ private def collectFiltered(file: File, glob: String): Seq[String] =
+ SupportsArchiveFormat.readArchiveEntries(
+ new Path(file.toURI), new Configuration(),
+ archivePathFilter = Some(new GlobPattern(glob))) { (entry, _) =>
+ Iterator.single(entry.getName)
+ }.toList
+
// ----- isArchivePath ------------------------------------------------------
test("isArchivePath: positive cases") {
@@ -306,7 +316,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
// HadoopFSUtils.shouldFilterOutPathName still apply -- mirroring a loose-file listing with
// the ignoredPathSegmentRegex option set to the same regex.
val entries = SupportsArchiveFormat.readArchiveEntries(
- new Path(tar.toURI), new Configuration(), Pattern.compile("(?!)")) { (entry, in) =>
+ new Path(tar.toURI), new Configuration(), Pattern.compile("(?!)"),
+ archivePathFilter = None) { (entry, in) =>
Iterator.single((entry.getName, new String(readAll(in), StandardCharsets.UTF_8)))
}.toList
assert(entries == Seq("_SUCCESS" -> "marker", "real.csv" -> "kept"))
@@ -321,7 +332,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
val opened = ArrayBuffer[String]()
// parseEntry yields a single element without reading the stream, so each invocation maps to
// exactly one consumed output element -- letting us observe when the next entry is opened.
- val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI), new Configuration()) {
+ val it = SupportsArchiveFormat.readArchiveEntries(
+ new Path(tar.toURI), new Configuration(), archivePathFilter = None) {
(entry, _) =>
opened += entry.getName
Iterator.single(entry.getName)
@@ -349,7 +361,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
val seen = ArrayBuffer[String]()
- val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI), new Configuration()) {
+ val it = SupportsArchiveFormat.readArchiveEntries(
+ new Path(tar.toURI), new Configuration(), archivePathFilter = None) {
(entry, in) =>
val body = new String(readAll(in), StandardCharsets.UTF_8)
in.close() // must NOT close the underlying archive
@@ -366,7 +379,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
val tar = new File(dir, "closeable.tar")
writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
- val it = SupportsArchiveFormat.readArchiveEntries(new Path(tar.toURI), new Configuration()) {
+ val it = SupportsArchiveFormat.readArchiveEntries(
+ new Path(tar.toURI), new Configuration(), archivePathFilter = None) {
(entry, _) => Iterator.single(entry.getName)
}
assert(it.hasNext)
@@ -395,7 +409,7 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
TaskContext.setTaskContext(ctx)
try {
val it = SupportsArchiveFormat.readArchiveEntries(
- new Path(tar.toURI), new Configuration()) {
+ new Path(tar.toURI), new Configuration(), archivePathFilter = None) {
(entry, _) => Iterator.single(entry.getName)
}
assert(it.hasNext)
@@ -464,7 +478,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
writeZip(zip, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b"), textEntry("c.csv", "c")))
val opened = ArrayBuffer[String]()
- val it = SupportsArchiveFormat.readArchiveEntries(new Path(zip.toURI), new Configuration()) {
+ val it = SupportsArchiveFormat.readArchiveEntries(
+ new Path(zip.toURI), new Configuration(), archivePathFilter = None) {
(entry, _) =>
opened += entry.getName
Iterator.single(entry.getName)
@@ -489,7 +504,8 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
writeZip(zip, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
val seen = ArrayBuffer[String]()
- val it = SupportsArchiveFormat.readArchiveEntries(new Path(zip.toURI), new Configuration()) {
+ val it = SupportsArchiveFormat.readArchiveEntries(
+ new Path(zip.toURI), new Configuration(), archivePathFilter = None) {
(entry, in) =>
val body = new String(readAll(in), StandardCharsets.UTF_8)
in.close() // must NOT close the underlying archive
@@ -607,4 +623,60 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
assert(collect(sevenZ) == Seq("real.csv" -> "kept"))
}
}
+
+ // ----- archivePathFilter ---------------------------------------------------
+
+ test("readArchiveEntries: archivePathFilter keeps only entries matching the glob") {
+ withTempDir { dir =>
+ val tar = new File(dir, "filter.tar")
+ writeTar(tar, Seq(
+ textEntry("sub/a.csv", "a"),
+ textEntry("sub/b.csv", "b"),
+ textEntry("other/c.csv", "c")))
+ // The glob matches the entry's full path, so `sub/*` selects only the `sub/` entries.
+ assert(collectFiltered(tar, "sub/*") == Seq("sub/a.csv", "sub/b.csv"))
+ }
+ }
+
+ test("readArchiveEntries: archivePathFilter with a `*` glob crosses directory boundaries") {
+ withTempDir { dir =>
+ val tar = new File(dir, "filter-ext.tar")
+ writeTar(tar, Seq(
+ textEntry("top.csv", "t"),
+ textEntry("sub/nested.csv", "n"),
+ textEntry("keep.json", "j")))
+ assert(collectFiltered(tar, "*.csv") == Seq("top.csv", "sub/nested.csv"))
+ }
+ }
+
+ test("readArchiveEntries: archivePathFilter matching nothing yields an empty iterator") {
+ withTempDir { dir =>
+ val tar = new File(dir, "filter-none.tar")
+ writeTar(tar, Seq(textEntry("a.csv", "a"), textEntry("b.csv", "b")))
+ assert(collectFiltered(tar, "nomatch/*").isEmpty)
+ }
+ }
+
+ test("readArchiveEntries: archivePathFilter applies on top of hidden-entry filtering") {
+ withTempDir { dir =>
+ val tar = new File(dir, "filter-hidden.tar")
+ writeTar(tar, Seq(
+ textEntry("data/real.csv", "kept"),
+ textEntry("data/_SUCCESS", "marker"))) // matches the glob but hidden by the default regex
+ assert(collectFiltered(tar, "data/*") == Seq("data/real.csv"))
+ }
+ }
+
+ test("archivePathFilter: an invalid glob is rejected with a clear error") {
+ val ex = intercept[IllegalArgumentException](
+ FileSourceOptions.compileArchivePathFilter("["))
+ assert(ex.getMessage.contains("archivePathFilter"))
+ }
+
+ test("archivePathFilter: an empty value disables the filter rather than matching nothing") {
+ val options = new FileSourceOptions(
+ Map(FileSourceOptions.ARCHIVE_PATH_FILTER -> ""))
+ assert(options.archivePathFilter.isEmpty)
+ assert(options.archivePathFilterPattern.isEmpty)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org