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 62db2d90ec8fd..1040bb0312875 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 c2a05a8c9d8b9..5988724070961 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 a1bf478f14f10..266e6ee835ced 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 ff25f957c29dc..dbfcff3f1fa5d 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 33de63072ef39..c0e50ee02e522 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 ddaf8ba2539ca..d2075a3304dfb 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 0bfee12d16664..388030dcf42df 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 42568a2e6b0a4..14e23a5fbdeec 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 4ede461a4d513..eeeb96d0e6e51 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 15d30aaab2a9a..c069abe3b4380 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 69f930600ae95..cf5544f923010 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 235aa428fbce1..a97cd7d57e418 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 c220dd5a957f9..62d2cd4a2eddc 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 14116279801d4..4ec07656f3f4b 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 26bf2cd3ccab6..7823efd6ae811 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 ab448a4f2c809..825c341e843da 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 c115f9bafa537..90c4b7765ba96 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 @@ -126,6 +126,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 0c362afd31d8e..d9eae971e975e 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) + } }