cloud-fan commented on code in PR #57808:
URL: https://github.com/apache/spark/pull/57808#discussion_r3728380094
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala:
##########
@@ -66,12 +70,34 @@ 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]] compiled once per JVM, so archive
reads reuse a single
Review Comment:
`lazy val` caches per `FileSourceOptions` instance, not per JVM. Please
describe the actual scope; the CSV, JSON, and XML inference paths also
construct matchers directly instead of sharing this value.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala:
##########
@@ -168,11 +171,19 @@ abstract class CSVDataSource extends Serializable with
Logging with SupportsArch
inputPaths: Seq[FileStatus],
parsedOptions: CSVOptions): StructType = {
val baseRdd = CSVDataSource.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
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) =>
+ SupportsArchiveFormat.readArchiveEntries(
+ path, stream.getConfiguration,
+ archivePathFilter =
+
archivePathFilterGlob.map(FileSourceOptions.compileArchivePathFilter)) {
Review Comment:
This compiles the same glob once per archive processed by the partition;
`JsonDataSource.scala:307` and `XmlDataSource.scala:389` do likewise. Please
use `mapPartitions` to compile once and reuse the matcher for every archive
stream in that partition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
