cloud-fan commented on code in PR #57388:
URL: https://github.com/apache/spark/pull/57388#discussion_r3624487048
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:
##########
@@ -526,29 +553,68 @@ object ParquetFileFormat extends Logging {
partFiles: Seq[FileStatus],
ignoreCorruptFiles: Boolean,
ignoreMissingFiles: Boolean = false): Seq[Footer] = {
+ val archiveEnabled =
SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
ThreadUtils.parmap(partFiles, "readingParquetFooters", 8,
preserveSparkThrowable = true) { currentFile =>
try {
// Skips row group information since we only need the schema.
// ParquetFileReader.readFooter throws RuntimeException, instead of
IOException,
// when it can't read the footer.
- Some(new Footer(currentFile.getPath,
- ParquetFooterReader.readFooter(
- HadoopInputFile.fromStatus(currentFile, conf), SKIP_ROW_GROUPS)))
+ 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)
+ } else {
+ Seq(new Footer(currentFile.getPath,
+ ParquetFooterReader.readFooter(
+ HadoopInputFile.fromStatus(currentFile, conf), SKIP_ROW_GROUPS)))
+ }
} catch {
case e: Exception if ignoreMissingFiles &&
ExceptionUtils.getThrowables(e).exists(_.isInstanceOf[FileNotFoundException]) =>
logWarning(log"Skipped missing file: ${MDC(PATH, currentFile)}", e)
- None
+ Seq.empty
case e: RuntimeException if ignoreCorruptFiles =>
logWarning(log"Skipped the footer in the corrupted file: ${MDC(PATH,
currentFile)}", e)
- None
+ Seq.empty
+ // A corrupt archive's failure arrives wrapped, not as a bare
RuntimeException; a missing
+ // file (FileNotFoundException) is excluded since ignoreMissingFiles
governs it.
+ case e: Exception if ignoreCorruptFiles && {
+ val root = Utils.getRootCause(e)
+ root.isInstanceOf[IOException] &&
!root.isInstanceOf[FileNotFoundException]
+ } =>
+ logWarning(log"Skipped the footer in the corrupted file: ${MDC(PATH,
currentFile)}", e)
+ Seq.empty
case e: Exception =>
throw
QueryExecutionErrors.cannotReadFooterForFileError(currentFile.getPath, e)
}
}.flatten
}
+ /** Reads every Parquet entry's footer in one archive. */
+ 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)
Review 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.
--
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]
