This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new be22910b7165 [SPARK-57591][SQL] Read and infer ORC schema from archives
be22910b7165 is described below
commit be22910b71650cab00d7883da3cc088ffb4de1fc
Author: akshatshenoi-db <[email protected]>
AuthorDate: Wed Aug 5 11:43:03 2026 +0800
[SPARK-57591][SQL] Read and infer ORC schema from archives
### What changes were proposed in this pull request?
This extends the archive-reader feature to the ORC data source, so a
`.tar`/`.tar.gz`/`.tgz`/`.zip`/`.7z` archive of ORC files is read as if it were
a directory of the entries it contains, gated by
`spark.sql.files.archive.reader.enabled` (default false). It follows the series
that added archive support to CSV (SPARK-57135, SPARK-57321), JSON
(SPARK-57419), text (SPARK-57478), XML (SPARK-57479), Avro (SPARK-57481), the
zip container (SPARK-57705), the `SupportsArchiveFormat` trait ext [...]
ORC is a random-access format like Parquet: a footer read needs the whole
file on disk, so it cannot be streamed entry-by-entry. It consumes the
random-access surface already on the `SupportsArchiveFormat` trait (added by
the Parquet port):
- `OrcFileFormat` mixes in `SupportsArchiveFormat`, returns
`archiveEntryFilter = true` (ORC part-files are often extensionless, so every
entry is read), reports `isSplitable = false` for an archive path, and in
`buildReaderWithPartitionValues` dispatches an archive input to
`readLocalizedEntries`, which unpacks each entry to a temp file and runs the
normal single-file reader over it.
- `OrcUtils` routes schema inference through the archive: the non-merge
path samples the first entry (`readArchiveSchemas(stopAtFirst = true)`,
matching ORC's existing sample-one behavior, SPARK-11500), and the
`mergeSchema` parallel path folds every entry (whole-archive-atomic on a
corrupt entry). A corrupt archive surfaces the same
`cannotReadFooterForFileError` a corrupt loose file does.
The shared test base is refactored so the localize-path tests
(vectorized-reader read parity, abandoned-read/corrupt-archive temp-dir
cleanup, extensionless entries, `mergeSchema` field union) live in
`ArchiveReadSuiteBase` behind `localizesEntries`, keyed on new
`archiveTempDirPrefix` / `vectorizedReaderConfKey` hooks, instead of being
duplicated per random-access format. `ParquetArchiveReadBase` is slimmed to set
those hooks; `OrcArchiveReadBase` reuses them.
### Why are the changes needed?
ORC was the remaining columnar format without archive-read support; this
brings it to parity with Parquet and the streaming formats, so users can read
archived ORC datasets without unpacking them first. Folding the shared
random-access tests into the base avoids duplicating them for every columnar
format.
### Does this PR introduce _any_ user-facing change?
Yes. When `spark.sql.files.archive.reader.enabled` is true (default false),
the ORC data source reads supported archives as directories of their entries,
for both scan and schema inference. With the flag off (the default) there is no
behavior change.
### How was this patch tested?
New `OrcTarArchiveReadSuite`, `OrcZipArchiveReadSuite`, and
`OrcSevenZArchiveReadSuite` run the shared `ArchiveReadSuiteBase` matrix over
ORC (read parity with a directory, schema inference, vectorized on/off,
corrupt/missing handling, `mergeSchema` union, temp-dir cleanup). The
refactored localize-path tests continue to run for Parquet via
`ParquetArchiveReadBase`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #57615 from akshatshenoi-db/archive-orc.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 4e4784459851299c99d24f09b09965be3b3084eb)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../execution/datasources/orc/OrcFileFormat.scala | 22 ++++-
.../sql/execution/datasources/orc/OrcUtils.scala | 97 ++++++++++++++++---
.../datasources/ArchiveReadSuiteBase.scala | 107 ++++++++++++++++++++-
.../execution/datasources/OrcArchiveReadBase.scala | 65 +++++++++++++
.../datasources/ParquetArchiveReadBase.scala | 90 +----------------
5 files changed, 281 insertions(+), 100 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
index bf0729bf4a0a..15d30aaab2a9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcFileFormat.scala
@@ -39,6 +39,7 @@ import
org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjectio
import org.apache.spark.sql.catalyst.types.DataTypeUtils.toAttributes
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.internal.SessionStateHelper
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
import org.apache.spark.util.{SerializableConfiguration, Utils}
@@ -50,8 +51,12 @@ class OrcFileFormat
extends FileFormat
with DataSourceRegister
with SessionStateHelper
+ with SupportsArchiveFormat
with Serializable {
+ // ORC part-files are often extensionless, so read every archive entry.
+ override protected def archiveEntryFilter(name: String): Boolean = true
+
override def shortName(): String = "orc"
override def toString: String = "ORC"
@@ -114,6 +119,10 @@ class OrcFileFormat
sparkSession: SparkSession,
options: Map[String, String],
path: Path): Boolean = {
+ val orcOptions = new OrcOptions(options, getSqlConf(sparkSession))
+ if (orcOptions.archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(path)) {
+ return false
+ }
true
}
@@ -168,8 +177,9 @@ class OrcFileFormat
SerializableConfiguration.broadcast(sparkSession.sparkContext,
hadoopConf)
val isCaseSensitive = sqlConf.caseSensitiveAnalysis
val orcFilterPushDown = sqlConf.orcFilterPushDown
+ val archiveFormatEnabled =
sqlConf.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
- (file: PartitionedFile) => {
+ def readSingleFile(file: PartitionedFile): Iterator[InternalRow] = {
val conf = broadcastedConf.value.value
val filePath = file.toPath
@@ -244,6 +254,16 @@ class OrcFileFormat
}
}
}
+
+ (file: PartitionedFile) => {
+ if (archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.toPath)) {
+ readLocalizedEntries(file, broadcastedConf.value.value, "orc-archive")
{ entryFile =>
+ readSingleFile(entryFile)
+ }
+ } else {
+ readSingleFile(file)
+ }
+ }
}
override def supportDataType(dataType: DataType): Boolean = dataType match {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
index e1676bd3ff51..69f930600ae9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcUtils.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources.orc
-import java.io.FileNotFoundException
+import java.io.{File, FileNotFoundException, IOException}
import java.nio.charset.StandardCharsets.UTF_8
import java.util.Locale
@@ -35,7 +35,7 @@ import org.apache.orc.{BooleanColumnStatistics,
ColumnStatistics, DateColumnStat
import org.apache.orc.mapred.{OrcInputFormat => OrcMapredInputFormat,
OrcStruct}
import org.apache.orc.mapreduce.OrcMapreduceRecordReader
-import org.apache.spark.{SPARK_VERSION_SHORT, SparkException}
+import org.apache.spark.{SPARK_VERSION_SHORT, SparkEnv, SparkException}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.PATH
@@ -47,12 +47,13 @@ import
org.apache.spark.sql.catalyst.parser.CatalystSqlParser
import org.apache.spark.sql.catalyst.util.{quoteIdentifier,
CaseInsensitiveMap, CharVarcharUtils}
import org.apache.spark.sql.connector.expressions.aggregate.{Aggregation,
Count, CountStar, Max, Min}
import org.apache.spark.sql.errors.QueryExecutionErrors
-import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils,
SchemaMergeUtils}
+import org.apache.spark.sql.execution.datasources.{AggregatePushDownUtils,
SchemaMergeUtils, SupportsArchiveFormat}
import org.apache.spark.sql.execution.datasources.orc.types.ops.OrcTypeOps
import org.apache.spark.sql.execution.datasources.v2.V2ColumnUtils
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
+import org.apache.spark.util.{ThreadUtils, Utils}
import org.apache.spark.util.ArrayImplicits._
-import org.apache.spark.util.ThreadUtils
object OrcUtils extends Logging {
@@ -184,11 +185,20 @@ object OrcUtils extends Logging {
: Option[StructType] = {
val ignoreCorruptFiles = new
FileSourceOptions(CaseInsensitiveMap(options)).ignoreCorruptFiles
val conf = sparkSession.sessionState.newHadoopConfWithOptions(options)
- files.iterator.map(file => readSchema(file.getPath, conf,
ignoreCorruptFiles)).collectFirst {
- case Some(schema) =>
- logDebug(s"Reading schema from file $files, got Hive schema string:
$schema")
- toCatalystSchema(schema)
- }
+ val ignoreMissingFiles =
+ new FileSourceOptions(CaseInsensitiveMap(options)).ignoreMissingFiles
+ val archiveFormatEnabled =
SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
+ files.iterator.flatMap { file =>
+ if (archiveFormatEnabled &&
SupportsArchiveFormat.isArchivePath(file.getPath)) {
+ readArchiveSchemas(conf, file, ignoreCorruptFiles, ignoreMissingFiles,
stopAtFirst = true)
+ .headOption
+ } else {
+ readSchema(file.getPath, conf, ignoreCorruptFiles).map { schema =>
+ logDebug(s"Reading schema from file $files, got Hive schema string:
$schema")
+ toCatalystSchema(schema)
+ }
+ }
+ }.collectFirst { case schema => schema }
}
/**
@@ -198,12 +208,77 @@ object OrcUtils extends Logging {
def readOrcSchemasInParallel(
files: Seq[FileStatus], conf: Configuration, ignoreCorruptFiles: Boolean,
ignoreMissingFiles: Boolean): Seq[StructType] = {
+ // Read outside `parmap`: its worker threads do not inherit the caller's
`SQLConf` thread-local,
+ // so `SQLConf.get` there would fall back to defaults and never take the
archive branch.
+ val archiveEnabled =
SQLConf.get.getConf(SQLConf.ARCHIVE_FORMAT_READER_ENABLED)
ThreadUtils.parmap(files, "readingOrcSchemas", 8) { currentFile =>
- OrcUtils.readSchema(currentFile.getPath, conf, ignoreCorruptFiles,
ignoreMissingFiles)
- .map(toCatalystSchema)
+ if (archiveEnabled &&
SupportsArchiveFormat.isArchivePath(currentFile.getPath)) {
+ readArchiveSchemas(conf, currentFile, ignoreCorruptFiles,
ignoreMissingFiles,
+ stopAtFirst = false)
+ } else {
+ OrcUtils.readSchema(currentFile.getPath, conf, ignoreCorruptFiles,
ignoreMissingFiles)
+ .map(toCatalystSchema).toSeq
+ }
}.flatten
}
+ /**
+ * Reads ORC entry schemas from one archive.
+ *
+ * @param stopAtFirst when true, returns just the first entry's schema
(sample-one); otherwise
+ * reads every entry so a corrupt entry fails the whole
archive.
+ */
+ private def readArchiveSchemas(
+ conf: Configuration,
+ archive: FileStatus,
+ ignoreCorruptFiles: Boolean,
+ ignoreMissingFiles: Boolean,
+ stopAtFirst: Boolean): Seq[StructType] = {
+ val tempDir = Utils.createTempDir(Utils.getLocalDir(SparkEnv.get.conf),
"orc-archive-infer")
+ // localizeEntries eagerly opens the first entry, so build it inside the
try; the finally must
+ // still delete tempDir when a corrupt archive throws there.
+ var entries: Iterator[(String, File)] = Iterator.empty
+ try {
+ entries = SupportsArchiveFormat.localizeEntries(archive.getPath, conf,
tempDir, _ => true)
+ // With ignore flags off, a corrupt entry throws to the per-archive
catch below. `.toList`
+ // reads every entry (whole archive atomic); `stopAtFirst` stays lazy
and stops at the first.
+ val schemas = entries.flatMap { case (_, entryFile) =>
+ try {
+ readSchema(new Path(entryFile.toURI), conf, ignoreCorruptFiles =
false,
+ ignoreMissingFiles = false).map(toCatalystSchema)
+ } finally entryFile.delete()
+ }
+ if (stopAtFirst) schemas.take(1).toList else schemas.toList
+ } catch {
+ // A corrupt container throws IOException at open; a corrupt entry
footer throws
+ // cannotReadFooterForFileError, whose root cause is also an IOException.
+ case e: Exception if ignoreMissingFiles &&
+
ExceptionUtils.getThrowables(e).exists(_.isInstanceOf[FileNotFoundException]) =>
+ logWarning(log"Skipped missing archive during inference: ${MDC(PATH,
archive.getPath)}", e)
+ Seq.empty
+ case e: Exception if {
+ val root = Utils.getRootCause(e)
+ root.isInstanceOf[IOException] &&
!root.isInstanceOf[FileNotFoundException]
+ } =>
+ if (ignoreCorruptFiles) {
+ logWarning(log"Skipped the corrupt archive during inference: " +
+ log"${MDC(PATH, archive.getPath)}", e)
+ Seq.empty
+ } else if (e.isInstanceOf[SparkException]) {
+ throw e // a corrupt entry footer already is
cannotReadFooterForFileError
+ } else {
+ // Match the loose-file footer error rather than leaking the raw
container IOException.
+ throw
QueryExecutionErrors.cannotReadFooterForFileError(archive.getPath, e)
+ }
+ } finally {
+ entries match {
+ case c: java.io.Closeable => c.close()
+ case _ =>
+ }
+ Utils.deleteRecursively(tempDir)
+ }
+ }
+
def inferSchema(sparkSession: SparkSession, files: Seq[FileStatus], options:
Map[String, String])
: Option[StructType] = {
val orcOptions = new OrcOptions(options, sparkSession.sessionState.conf)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
index 943eb637e8fa..ab448a4f2c80 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ArchiveReadSuiteBase.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.{DataFrame, QueryTest, Row}
import org.apache.spark.sql.functions.{col, input_file_name, struct}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{StringType, StructType}
+import org.apache.spark.sql.types.{DataType, StringType, StructType}
import org.apache.spark.util.Utils
/**
@@ -183,6 +183,19 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
*/
protected def localizesEntries: Boolean = false
+ /**
+ * Prefix of the per-entry temp dir a [[localizesEntries]] format unpacks
each entry into (e.g.
+ * `parquet-archive`); its inference variant is `<prefix>-infer`. Gates the
shared temp-dir
+ * cleanup tests.
+ */
+ protected def archiveTempDirPrefix: String = ""
+
+ /**
+ * The config key toggling this format's vectorized/columnar reader, if it
has one. When set, the
+ * shared localize-path read test runs with the reader both on and off.
+ */
+ protected def vectorizedReaderConfKey: Option[String] = None
+
/** Sample data with a nested struct column, used by the complex-type test.
*/
protected def complexSampleDf: DataFrame =
Seq((1, "NYC", "10001"), (2, "SF", "94105")).toDF("id", "city", "zip")
@@ -530,6 +543,98 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
}
}
}
+
+ // Temp dirs (of `prefix`) surviving under the executor-local dir; the
read/inference variants
+ // must be removed on task completion, so `after -- before` catches a leak.
+ def archiveTempDirs(prefix: String): Set[String] = {
+ val localDir = new File(Utils.getLocalDir(spark.sparkContext.getConf))
+ Option(localDir.listFiles()).getOrElse(Array.empty)
+ .filter(_.getName.startsWith(prefix)).map(_.getName).toSet
+ }
+
+ val vectorizedCases =
vectorizedReaderConfKey.map(Seq(_)).getOrElse(Seq.empty)
+ .flatMap(key => Seq(true, false).map(key -> _))
+ (if (vectorizedCases.nonEmpty) vectorizedCases else Seq("" ->
false)).foreach {
+ case (confKey, vectorized) =>
+ val label = if (confKey.isEmpty) "" else s" with vectorized reader =
$vectorized"
+ test(s"archive reads return the same rows$label") {
+ val conf = if (confKey.isEmpty) Map.empty[String, String]
+ else Map(confKey -> vectorized.toString)
+ withSQLConf(conf.toSeq: _*) {
+ assertArchiveMatchesDir(
+ Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice"), (2,
"Bob")))))
+ }
+ }
+ }
+
+ test("an abandoned read (LIMIT) over an archive returns partial rows and
cleans up") {
+ withArchiveFile() { archive =>
+ val parts = (0 until 4).map(i => entryName(i) ->
encodeFile(sampleDf((i, s"v$i"))))
+ writeArchive(archive, parts)
+ val before = archiveTempDirs(archiveTempDirPrefix)
+ assert(read(archive.getCanonicalPath).limit(2).collect().length == 2)
+ assert((archiveTempDirs(archiveTempDirPrefix) -- before).isEmpty,
+ "the read's temp dir was not cleaned up")
+ }
+ }
+
+ test("extensionless entries are read and inferred like a directory of
part-files") {
+ val data = sampleDf((1, "Alice"), (2, "Bob"))
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq("part-00000" -> encodeFile(data)))
+ checkAnswer(read(archive.getCanonicalPath), data)
+ assert(inferredSchema(Seq(archive.getCanonicalPath)).fieldNames.toSet
== Set("id", "name"),
+ "an extensionless entry should be inferred like a directory of
part-files")
+ }
+ }
+
+ test("a corrupt archive cleans up its read temp dir rather than leaking
it") {
+ // A corrupt archive throws before the read returns an iterator, but
must not leak its dir.
+ withArchiveFile(corruptArchiveExtension) { archive =>
+ writeCorruptArchive(archive)
+ val before = archiveTempDirs(archiveTempDirPrefix)
+ intercept[SparkException](read(archive.getCanonicalPath).collect())
+ assert((archiveTempDirs(archiveTempDirPrefix) -- before).isEmpty,
+ "a corrupt archive leaked its read temp dir")
+ }
+ }
+
+ test("a corrupt archive cleans up its inference temp dir rather than
leaking it") {
+ // Inference localizes entries too, on a worker without a TaskContext; a
corrupt archive
+ // throws during that eager localize and must not leak the inference
temp dir.
+ withArchiveFile(corruptArchiveExtension) { archive =>
+ writeCorruptArchive(archive)
+ val before = archiveTempDirs(s"$archiveTempDirPrefix-infer")
+
intercept[SparkException](inferredSchema(Seq(archive.getCanonicalPath)))
+ assert((archiveTempDirs(s"$archiveTempDirPrefix-infer") --
before).isEmpty,
+ "a corrupt archive leaked its inference temp dir")
+ }
+ }
+
+ test("archive inference unions differing fields across entries with
mergeSchema=true") {
+ // mergeSchema=true folds every entry's schema; over an archive, one
unpacked entry at a time.
+ // Compare field name/type pairs as a set rather than by exact
StructType equality: field
+ // order across archive entries follows the parallel merge order and is
not a guaranteed
+ // contract, and some formats (ORC) merge entries unordered. Only the
name/type union matters.
+ val withName = sampleDf((1, "Alice"), (2, "Bob"))
+ val idExtra = Seq((3, 30)).toDF("id", "extra")
+ val entries = Seq(entryName(0) -> encodeFile(withName), entryName(1) ->
encodeFile(idExtra))
+ val merge = Map("mergeSchema" -> "true")
+ def fieldTypes(s: StructType): Set[(String, DataType)] =
+ s.fields.map(f => f.name -> f.dataType).toSet
+ withArchiveFile() { archive =>
+ writeArchive(archive, entries)
+ val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath),
merge)
+ withTempDir { dir =>
+ entries.foreach { case (n, b) => Files.write(new File(dir,
n).toPath, b) }
+ assert(archiveSchema.fieldNames.toSet == Set("id", "name", "extra"),
+ s"expected the union of entry fields, got $archiveSchema")
+ assert(fieldTypes(archiveSchema) ==
+ fieldTypes(inferredSchema(Seq(dir.getCanonicalPath), merge)),
+ s"archive mergeSchema inference diverged from a directory read;
got $archiveSchema")
+ }
+ }
+ }
}
// ----- shared complex-type test (run when `supportsComplexTypes`)
----------
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/OrcArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/OrcArchiveReadBase.scala
new file mode 100644
index 000000000000..a63152caa975
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/OrcArchiveReadBase.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s hooks to ORC (entries unpacked to a local
file for footer
+ * random access). ORC is self-describing, so the base's schema-inference
tests run too.
+ */
+trait OrcArchiveReadBase extends ArchiveReadSuiteBase {
+
+ override protected def format: String = "orc"
+
+ override protected def fileExtension: String = "orc"
+
+ override protected def readOptions: Map[String, String] = Map.empty
+
+ override protected def readSchema: String = "id INT, name STRING"
+
+ // ORC has authoritative per-file schemas and only unions under
`mergeSchema`, so it opts out of
+ // the by-name default-inference union (covered by the shared localize-path
tests instead).
+ override protected def supportsSchemaMerge: Boolean = false
+
+ // ORC samples one part-file for non-merge inference.
+ override protected def inferenceSamplesOneFile: Boolean = true
+
+ // ORC unpacks each entry to a local temp file for footer random access.
+ override protected def localizesEntries: Boolean = true
+
+ override protected def archiveTempDirPrefix: String = "orc-archive"
+
+ override protected def vectorizedReaderConfKey: Option[String] =
+ Some(SQLConf.ORC_VECTORIZED_READER_ENABLED.key)
+}
+
+class OrcTarArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with OrcArchiveReadBase
+ with TarArchiveReadBase
+
+class OrcZipArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with OrcArchiveReadBase
+ with ZipArchiveReadBase
+
+class OrcSevenZArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with OrcArchiveReadBase
+ with SevenZArchiveReadBase
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala
index c09f764ba3d2..374098789f16 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/ParquetArchiveReadBase.scala
@@ -18,15 +18,12 @@
package org.apache.spark.sql.execution.datasources
import java.io.File
-import java.nio.file.Files
import org.apache.hadoop.fs.{FileStatus, Path}
-import org.apache.spark.SparkException
import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StructType
-import org.apache.spark.util.Utils
/**
* Binds [[ArchiveReadSuiteBase]]'s hooks to Parquet (entries unpacked to a
local file for footer
@@ -34,8 +31,6 @@ import org.apache.spark.util.Utils
*/
trait ParquetArchiveReadBase extends ArchiveReadSuiteBase {
- import testImplicits._
-
override protected def format: String = "parquet"
override protected def fileExtension: String = "parquet"
@@ -54,89 +49,10 @@ trait ParquetArchiveReadBase extends ArchiveReadSuiteBase {
// Parquet unpacks each entry to a local temp file for footer random access.
override protected def localizesEntries: Boolean = true
- for (vectorized <- Seq(true, false)) {
- test(s"archive reads return the same rows with vectorized reader =
$vectorized") {
- withSQLConf(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key ->
vectorized.toString) {
- assertArchiveMatchesDir(
- Seq(entryName(0) -> encodeFile(sampleDf((1, "Alice"), (2, "Bob")))))
- }
- }
- }
-
- test("an abandoned read (LIMIT) over an archive returns partial rows and
cleans up") {
- def archiveTempDirs(localDir: File): Set[String] =
- Option(localDir.listFiles()).getOrElse(Array.empty)
- .filter(_.getName.startsWith("parquet-archive")).map(_.getName).toSet
- withArchiveFile() { archive =>
- val parts = (0 until 4).map(i => entryName(i) -> encodeFile(sampleDf((i,
s"v$i"))))
- writeArchive(archive, parts)
- val localDir = new File(Utils.getLocalDir(spark.sparkContext.getConf))
- val before = archiveTempDirs(localDir)
- assert(read(archive.getCanonicalPath).limit(2).collect().length == 2)
- // This read's per-entry temp dir (prefix `parquet-archive`) must be
removed on task
- // completion, so no new one survives.
- assert((archiveTempDirs(localDir) -- before).isEmpty,
- "the read's temp dir was not cleaned up")
- }
- }
-
- test("extensionless entries are read and inferred like a directory of
part-files") {
- val data = sampleDf((1, "Alice"), (2, "Bob"))
- withArchiveFile() { archive =>
- writeArchive(archive, Seq("part-00000" -> encodeFile(data)))
- checkAnswer(read(archive.getCanonicalPath), data)
- assert(inferredSchema(Seq(archive.getCanonicalPath)).fieldNames.toSet ==
Set("id", "name"),
- "an extensionless entry should be inferred like a directory of
part-files")
- }
- }
+ override protected def archiveTempDirPrefix: String = "parquet-archive"
- private def parquetArchiveTempDirs(prefix: String): Set[String] = {
- val localDir = new File(Utils.getLocalDir(spark.sparkContext.getConf))
- Option(localDir.listFiles()).getOrElse(Array.empty)
- .filter(_.getName.startsWith(prefix)).map(_.getName).toSet
- }
-
- test("a corrupt archive cleans up its read temp dir rather than leaking it")
{
- // A corrupt archive throws before the read returns an iterator, but must
not leak the temp dir.
- withArchiveFile(corruptArchiveExtension) { archive =>
- writeCorruptArchive(archive)
- val before = parquetArchiveTempDirs("parquet-archive")
- intercept[SparkException](read(archive.getCanonicalPath).collect())
- assert((parquetArchiveTempDirs("parquet-archive") -- before).isEmpty,
- "a corrupt archive leaked its read temp dir")
- }
- }
-
- test("a corrupt archive cleans up its inference temp dir rather than leaking
it") {
- // Inference localizes entries too (readArchiveFooters), on a worker
without a TaskContext; a
- // corrupt archive throws during that eager localize and must not leak
parquet-archive-infer.
- withArchiveFile(corruptArchiveExtension) { archive =>
- writeCorruptArchive(archive)
- val before = parquetArchiveTempDirs("parquet-archive-infer")
- intercept[SparkException](inferredSchema(Seq(archive.getCanonicalPath)))
- assert((parquetArchiveTempDirs("parquet-archive-infer") --
before).isEmpty,
- "a corrupt archive leaked its inference temp dir")
- }
- }
-
- test("archive inference unions differing fields across entries with
mergeSchema=true") {
- // mergeSchema=true folds every entry's footer; over an archive, one
unpacked entry at a time.
- val withName = sampleDf((1, "Alice"), (2, "Bob"))
- val idExtra = Seq((3, 30)).toDF("id", "extra")
- val entries = Seq(entryName(0) -> encodeFile(withName), entryName(1) ->
encodeFile(idExtra))
- val merge = Map("mergeSchema" -> "true")
- withArchiveFile() { archive =>
- writeArchive(archive, entries)
- val archiveSchema = inferredSchema(Seq(archive.getCanonicalPath), merge)
- withTempDir { dir =>
- entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath,
b) }
- assert(archiveSchema.fieldNames.toSet == Set("id", "name", "extra"),
- s"expected the union of entry fields, got $archiveSchema")
- assert(archiveSchema == inferredSchema(Seq(dir.getCanonicalPath),
merge),
- s"archive mergeSchema inference diverged from a directory read; got
$archiveSchema")
- }
- }
- }
+ override protected def vectorizedReaderConfKey: Option[String] =
+ Some(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key)
test("inference skips a missing archive among good ones
(ignoreMissingFiles)") {
// Exercised on ParquetFileFormat.inferSchema(files) directly: inference
now runs on the
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
