cloud-fan commented on code in PR #57388:
URL: https://github.com/apache/spark/pull/57388#discussion_r3622376230
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala:
##########
@@ -445,6 +472,10 @@ class ParquetFileFormat
}
object ParquetFileFormat extends Logging {
+
+ private[parquet] def isParquetArchiveEntry(name: String): Boolean =
+ name.toLowerCase(Locale.ROOT).endsWith(".parquet")
Review 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.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormat.scala:
##########
@@ -88,10 +90,116 @@ trait SupportsArchiveFormat extends Logging {
}
}
+ /** Which archive entries are data files of this format; others are skipped.
*/
+ protected def archiveEntryFilter(name: String): Boolean =
+ throw new UnsupportedOperationException(
+ s"${getClass.getName} does not support random-access archive reads")
+
+ /**
+ * 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.
+ *
+ * @param path the archive path
+ * @param conf Hadoop configuration used to open the archive
+ * @param localDir directory the per-entry temp files are created under
+ * @return an iterator of `(entryName, localFile)` for the kept entries, in
archive order
+ */
+ private def localizeEntries(
+ path: Path,
+ conf: Configuration,
+ localDir: File): Iterator[(String, File)] =
+ SupportsArchiveFormat.localizeEntries(path, conf, localDir,
archiveEntryFilter)
+
+ /**
+ * Reads an archive by unpacking each entry to a temp file and applying
`readEntry`, for a
+ * format that needs a complete file on disk (random access).
+ *
+ * @param file the archive as a [[PartitionedFile]]
+ * @param conf Hadoop configuration used to open the archive
+ * @param tempPrefix prefix for the per-task temp dir the entries are
unpacked into
+ * @param readEntry reads one unpacked entry file into rows
+ * @return iterator of rows across all entries
+ */
+ protected def readLocalizedEntries(
+ file: PartitionedFile,
+ conf: Configuration,
+ 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)
Review 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.
--
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]
