[SPARK-58609][SQL] Add archivePathFilter option to select inner archive entries - #57808
[SPARK-58609][SQL] Add archivePathFilter option to select inner archive entries#57808akshatshenoi-db wants to merge 2 commits into
Conversation
…ve entries 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. `pathGlobFilter` cannot express this: it must already match the archive file for the file to be listed, so it cannot simultaneously select a subset of the entries inside the archive. The glob is applied in addition to `ignoredPathSegmentRegex` (both must pass), and is inert when archive reading is disabled. `FileSourceOptions` carries the glob string plus `archivePathFilterPattern`, a transient lazy val holding the compiled Hadoop `GlobPattern`, so the glob is compiled once per executor JVM rather than once per archive. The glob string is validated on the driver so an invalid glob fails fast. `SupportsArchiveFormat.readArchiveEntries` and the random-access `readLocalizedEntries` / `localizeEntries` take an optional filter (default `None`, so existing callers are unaffected) applied in `shouldSkipEntry`. It is threaded from the read paths of CSV, JSON, text, XML, Avro, binaryFile, Parquet and ORC.
…ply it to inference - An empty `archivePathFilter` was retained and compiled to a `GlobPattern` that matches only the empty string, so every archive entry was skipped and the scan returned zero rows. Treat an empty value as absent, matching how an empty `ignoredPathSegmentRegex` disables that filter. - Remove the `None` default from the `archivePathFilter` parameter of `readArchiveEntries`, `readLocalizedEntries` and `localizeEntries`, so no call site can silently skip it. - Apply the filter to schema inference, which previously read every entry while the scan honored the filter. The inferred schema was therefore computed over a superset of the scanned entries. Threaded through CSV/JSON/XML (capturing the glob string, since the compiled `GlobPattern` is not serializable), Avro, ORC and Parquet. - Add a shared `ArchiveReadSuiteBase` case that puts an extra column in a filtered-out entry and asserts the inferred schema excludes it, so it runs for every format and container.
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
0 blocking, 1 non-blocking, 1 nit.
The filtering behavior and cross-format propagation are coherent, with two non-blocking cleanup items around matcher cache scope and repeated inference-time compilation.
Nits: 1 minor item (see inline comments).
Suggestions (1)
- sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala:185: This compilation runs once per archive in the flatMap callback, so partitions containing multiple archives repeatedly compile the same glob; JSON and XML inference have the same shape. Please compile once in mapPartitions and reuse the matcher across that partition's archive streams. -- see inline
Verification
I traced the option from FileSourceOptions through the shared archive-entry predicate and through scan and schema-inference paths for the supported formats. The filter is applied before parsing and composes with the existing per-segment hidden-path check; Parquet and ORC transport the string through Hadoop configuration for parallel schema work, while CSV, JSON, and XML currently compile it inside each archive callback.
PR metadata suggestions
- Correct the claim that archivePathFilterPattern is compiled once per executor JVM; it is cached per FileSourceOptions instance, and CSV/JSON/XML inference currently compiles once per archive.
| } | ||
|
|
||
| /** | ||
| * The effective [[archivePathFilter]] compiled once per JVM, so archive reads reuse a single |
There was a problem hiding this 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.
| SupportsArchiveFormat.readArchiveEntries( | ||
| path, stream.getConfiguration, | ||
| archivePathFilter = | ||
| archivePathFilterGlob.map(FileSourceOptions.compileArchivePathFilter)) { |
There was a problem hiding this 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.
2 participants
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 likesubdir/*and*/*.csvwork.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:
FileSourceOptionsgains thearchivePathFilteroption (the glob string) plusarchivePathFilterPattern, a@transient lazy valholding the compiled HadoopGlobPattern. Keeping the compiled matcher on the options object means the glob is compiled once per executor JVM rather than once per archive; it istransientbecauseGlobPatternis not serializable, so the string travels and executors recompile on first use. The glob string is validated on the driver, so an invalid glob raises a clearIllegalArgumentExceptionnaming the option.SupportsArchiveFormat.readArchiveEntriesand the random-accessreadLocalizedEntries/localizeEntriestake an optional filter (defaultNone, so existing callers are unaffected), applied inshouldSkipEntry.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?
pathGlobFiltercannot 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. WithoutarchivePathFilterthere 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 byspark.sql.files.archive.reader.enabled(defaultfalse).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:sub/*),ignoredPathSegmentRegex(a hidden entry matching the glob is still skipped).New unit cases in
SupportsArchiveFormatSuitecover 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