[SPARK-57590][SQL] Read and infer Parquet schema from archives#57388
[SPARK-57590][SQL] Read and infer Parquet schema from archives#57388akshatshenoi-db wants to merge 8 commits into
Conversation
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
2 blocking, 0 non-blocking, 0 nits.
The overall approach fits the existing archive abstraction, but two blocking edge cases can silently omit valid data or retain task-local files after initialization failures.
Design / architecture (1)
- sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:477: Archive entry selection is stricter than normal Parquet file selection and silently drops valid extensionless files. -- see inline
Correctness (1)
- sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:129: A failure while opening or copying the first entry leaks the new temp directory until executor shutdown. -- see inline
Verification
Reviewed all seven changed-file artifacts and traced archive localization through readArchiveEntries, FileScanRDD, normal Parquet file-index selection, footer inference, and Utils.createTempDir. Mechanical link, text-quality, and contract scanners completed with exact-manifest coverage. Tests were not run as part of this review.
| object ParquetFileFormat extends Logging { | ||
|
|
||
| private[parquet] def isParquetArchiveEntry(name: String): Boolean = | ||
| name.toLowerCase(Locale.ROOT).endsWith(".parquet") |
There was a problem hiding this comment.
Please remove the .parquet suffix requirement. Normal Parquet directory scans accept any visible, non-metadata filename, so an archive entry such as part-00000 is silently omitted even though the same file is readable after unpacking. The shared archive filtering already handles hidden paths; an extensionless-entry test should cover both reading and inference.
| tempPrefix: String)( | ||
| readEntry: PartitionedFile => Iterator[InternalRow]): Iterator[InternalRow] = { | ||
| val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), tempPrefix) | ||
| val entries = localizeEntries(file.toPath, conf, tempDir) |
There was a problem hiding this comment.
Register task cleanup before constructing entries, and delete tempDir if construction throws. readArchiveEntries eagerly probes its iterator, so a bad archive or a failure copying the first entry escapes here before the listener is installed and before FileScanRDD receives anything it can close. That partial directory then remains until executor shutdown; please cover this with a corrupt or truncated first-entry cleanup test.
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
2 addressed, 0 remaining, 1 new. (1 newly introduced, 0 late catches.)
1 blocking, 0 non-blocking, 0 nits.
Correctness (1)
- sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:598: Put
localizeEntriesinside thetry(or register task cleanup before calling it). That helper eagerly opens and copies the first archive entry, so a corrupt archive or first-entry copy failure throws here before thefinallystarts; the newly createdparquet-archive-inferdirectory is then left on the executor. The new corrupt-archive cleanup test covers the scan reader, but the corrupt inference tests do not check this separate temp directory. -- see inline
Verification
Reviewed all seven changed artifacts and traced localization through readArchiveEntries, the scan reader, and distributed schema inference. The three prepared mechanical scanners completed with exact-manifest coverage. Tests were not run as part of this review.
| private def readArchiveFooters(conf: Configuration, archive: FileStatus): Seq[Footer] = { | ||
| val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf), "parquet-archive-infer") | ||
| // No TaskContext here, so close the archive stream deterministically rather than on task end. | ||
| val entries = SupportsArchiveFormat.localizeEntries(archive.getPath, conf, tempDir, _ => true) |
There was a problem hiding this comment.
Move this call inside the try so initialization failures also reach temp-directory cleanup. localizeEntries eagerly opens and copies the first entry; if a corrupt archive or copy failure throws here, the finally never starts and the new parquet-archive-infer directory remains on the executor. Please add inference-specific cleanup coverage as well.
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
1 addressed, 0 remaining, 1 new. (0 newly introduced, 1 late catch.)
0 blocking, 1 non-blocking, 0 nits.
Correctness (1)
- sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:100: The localization helper overstates its one-file-on-disk contract; the iterator relies on each caller to delete prior results before advancing. -- see inline
Verification
Reviewed all seven changed artifacts and traced scan iteration, reader closure, task cleanup, distributed footer inference, corrupt/missing-file handling, and the archive config gate. Mechanical link, text-quality, and contract scanners completed with exact-manifest coverage. Tests were not run as part of this review.
|
|
||
| /** | ||
| * Materializes each kept entry (those passing [[archiveEntryFilter]]) to a temp file under | ||
| * `localDir`, lazily one at a time, so only one entry occupies disk at once. |
There was a problem hiding this comment.
Please scope this guarantee to readLocalizedEntries, or document that callers must delete each returned file before advancing. localizeEntries creates a new temp file on every next() and does not remove earlier results, so the helper itself does not ensure that only one entry occupies disk.
…hives Adds Parquet to the archive-reader feature (SPARK-57135 CSV, SPARK-57321 CSV-infer, SPARK-57419 JSON, SPARK-57478 text, SPARK-57479 XML, SPARK-57481 Avro, SPARK-57705 zip container, SPARK-58110 SupportsArchiveFormat trait). Parquet is the first random-access archive format: unlike the streaming formats, it needs its trailing footer, so archive entries are unpacked to a local temp file one at a time rather than streamed. This adds the random-access localization surface to the SupportsArchiveFormat trait (readLocalizedEntries / archiveEntryFilter, plus companion copyEntryToLocalFile / localizeEntries), which was deferred from SPARK-58110 until its first consumer. ParquetFileFormat mixes in the trait, marks archives non-splitable, dispatches archive reads through readLocalizedEntries reusing the plain per-file reader, and folds per-entry footer reads into readParquetFootersInParallel so schema inference expands archives. Gated by spark.sql.files.archive.reader.enabled (default false); loose Parquet reads are unaffected.
… for scan-all-inputs formats The 'archive inference widens a column's type across entries' test was gated on supportsSchemaMerge, which dropped it for CSV (merge off, but it scans all inputs and widens by inference). Gate it on !inferenceSamplesOneFile instead: it stays in the schema-inference block for CSV/text and is excluded only for Parquet, which samples one file and can't widen across entries.
… temp-dir/stream leaks - Archive entry selection no longer requires a .parquet suffix. Normal Parquet directory scans read any visible non-metadata file, so entries like part-00000 were silently omitted; archiveEntryFilter now accepts every entry (the shared filter already drops directories and hidden/metadata paths). - readLocalizedEntries registers its temp-dir cleanup listener before probing the first entry and deletes the temp dir if construction throws, so a corrupt archive or a first-entry copy failure -- which escapes before FileScanRDD receives a closeable iterator -- no longer leaks the directory. - readArchiveFooters closes the archive stream in a finally rather than relying on a task-completion listener: it runs on a ThreadUtils.parmap worker with no TaskContext, so a footer read that throws left the stream open (the leaked file stream that aborted ParquetTar/ZipArchiveReadSuite in CI). - Adds an extensionless-entry read+inference test and a corrupt-archive temp-dir cleanup test.
…leanup runs readArchiveFooters built the localizeEntries iterator before the try, so a corrupt archive throwing during the eager first-entry localize skipped the finally and leaked the parquet-archive-infer temp dir. Move the localize inside the try and add an inference-path temp-dir cleanup test.
…disk note to the caller The private localizeEntries writes a new temp file per next() and never deletes earlier ones, so it does not itself guarantee one entry on disk at a time -- readLocalizedEntries does, by deleting each file before advancing. Reword the Scaladoc to put that responsibility on the caller.
7z landed as a container (SPARK-58246), and Parquet's archive read is container-agnostic (readLocalizedEntries -> localizeEntries -> readArchiveEntries -> openArchiveStream, which now handles .7z), so Parquet reads 7z archives with no production change. Adds ParquetSevenZArchiveReadSuite and, matching the post-7z test layout on master, folds the per-container Parquet leaf suites into ParquetArchiveReadBase (removing the standalone Tar/Zip suite files).
3dbe318 to
cd05724
Compare
akshatshenoi-db
changed the title
2 participants
What changes were proposed in this pull request?
This extends the tar/zip archive-reader feature to Parquet, continuing the series
(SPARK-57135 CSV, SPARK-57321 CSV inference, SPARK-57419 JSON, SPARK-57478 text,
SPARK-57479 XML, SPARK-57481 Avro, SPARK-57705 zip container, and SPARK-58110 which
extracted the shared
SupportsArchiveFormattrait). With the feature enabled, anarchive (
.tar/.tar.gz/.tgz/.zip) of Parquet files is read like a directory ofits entries.
Parquet is the first random-access archive format. The streaming formats parse each
entry straight from an
InputStream, but Parquet needs its trailing footer, so anarchive entry cannot be streamed. This PR:
SupportsArchiveFormat: the traitgains
readLocalizedEntries(unpack each kept entry to a temp file, one at a time,read it, delete it) and the
archiveEntryFilterhook; the companion gainscopyEntryToLocalFileandlocalizeEntries(so an executor-side inference path canlocalize without a trait instance). This is the surface that was deferred from
SPARK-58110 until its first consumer landed - this is that consumer.
ParquetFileFormatmixes in the trait, returnsisSplitable = falsefor an archive(the archive is one split), dispatches an archive read through
readLocalizedEntriesreusing the existing plain per-file reader, and folds per-entry footer reads into
readParquetFootersInParallelso schema inference expands archives. A corrupt entrycondemns its whole archive during inference (matching the whole-file semantics of a
corrupt loose Parquet file), while a corrupt entry during a read skips the rest of
that archive as the streaming formats do.
input_file_name()/_metadata.file_pathreport the archive path, not a per-entrytemp path.
The SPARK-37089 use-after-free guard is preserved: the per-task completion listener
deletes only the temp dir;
FileScanRDDcloses the per-entry reader so the vectorizedreader's off-heap column vectors are not freed while still in use.
This also fixes a stale doc reference left by SPARK-58110 in
SupportsArchiveFormatSuite(the comment pointed at the removedZipArchiveReader'sscaladoc).
Why are the changes needed?
Parquet is a common archived format, and the archive-reader feature already supports the
text-oriented formats and Avro. Extending it to Parquet lets users ingest tarred/zipped
Parquet without unpacking first, and establishes the random-access path that ORC and
other footer-based formats will reuse.
Does this PR introduce any user-facing change?
Yes, gated by
spark.sql.files.archive.reader.enabled(defaultfalse). When enabled,reading a path that is a supported archive of Parquet files returns the rows of its
entries; when disabled (the default), behavior is unchanged. Loose Parquet reads are
unaffected either way.
How was this patch tested?
Added
ParquetTarArchiveReadSuiteandParquetZipArchiveReadSuite(each binding the newParquetArchiveReadBaseto the tar and zip harnesses), covering vectorized androw-based reads, schema inference (including
mergeSchemaunion and type widening acrossentries),
input_file_name/_metadataparity with the archive path, per-entry temp-filecleanup on an abandoned (
LIMIT) read, andignoreCorruptFiles/ignoreMissingFilesbehavior for both read and inference. Extended the shared
ArchiveReadSuiteBasewith aninferenceSamplesOneFilehook (Parquet samples one part-file per SPARK-11500) and aninput_file_nameparity test.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)