You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/05 03:43:20 UTC
(spark) branch master updated: [SPARK-57591][SQL] Read and infer ORC schema from archives
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 4e4784459851 [SPARK-57591][SQL] Read and infer ORC schema from archives
4e4784459851 is described below
commit 4e4784459851299c99d24f09b09965be3b3084eb
Author: akshatshenoi-db <ak...@databricks.com>
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 <ak...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org