cloud-fan commented on code in PR #57808:
URL: https://github.com/apache/spark/pull/57808#discussion_r3730628524
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala:
##########
@@ -168,29 +171,39 @@ 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 once per partition and reuse for every archive it holds.
+ val archivePathFilter =
archivePathFilterGlob.map(FileSourceOptions.compileArchivePathFilter)
Review Comment:
Could this be a partition-local `lazy val`? As written, every partition
compiles the glob before inspecting its streams, including partitions
containing only ordinary files. Lazy initialization would preserve
once-per-partition reuse while compiling only if the archive branch is reached.
The same applies to `JsonDataSource.scala:302` and `XmlDataSource.scala:384`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/FileSourceOptions.scala:
##########
@@ -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]] compiled once per instance of this
class, so the archive
Review Comment:
```suggestion
* The effective [[archivePathFilter]] is compiled once per instance of
this class, so the archive
```
##########
.isaac/config.json:
##########
@@ -0,0 +1,3 @@
+{
+ "sync_reminder_last_shown": "2026-08-06"
Review Comment:
Please remove this file from the PR. It is local Isaac reminder state and is
unrelated to the archive reader change.
--
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]
