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 1ff805b419c7 [SPARK-57481][SQL] Read and infer Avro schema from tar
archives
1ff805b419c7 is described below
commit 1ff805b419c7e8280c09bab14e6032fec815a562
Author: akshatshenoi-db <[email protected]>
AuthorDate: Wed Jun 24 13:10:12 2026 -0700
[SPARK-57481][SQL] Read and infer Avro schema from tar archives
### What changes were proposed in this pull request?
Extend the Avro data source to read and infer schema from
`.tar`/`.tar.gz`/`.tgz` archives when `spark.sql.files.archive.reader.enabled`
is set, continuing the tar-archive reader series: SPARK-57135 (CSV read),
SPARK-57321 (CSV inference), SPARK-57419 (JSON), SPARK-57478 (text),
SPARK-57479 (XML).
Reads stream each archive entry through a forward-only `DataFileStream`,
deserializing it like a standalone Avro file. The archive is never unpacked to
disk and memory stays bounded. Inference reads each entry's writer schema from
its Avro header (records are never scanned), also streamed, and uses the first
readable schema like a directory read -- Avro does not merge schemas across
files. The whole archive is a single split (see `isSplitable`), so the dispatch
lives on the V1 read pa [...]
On the test side, `ArchiveReadSuiteBase.encodeFile` becomes a shared
default (the format's single part-file bytes via `readOptions ++
writeOptions`), so the CSV/JSON/XML traits drop their identical overrides.
Because Avro infers from an embedded header rather than from content, it runs
the inference-parity tests but excludes the type-widening test and opts out of
the schema-merge tests, and it adds two streaming-reader regression tests (a
truncated entry must fail fast, a drip-fed ent [...]
### Why are the changes needed?
The archive reader already supports CSV, JSON, text, and XML. Avro is a
common container for archived data, and extending the same opt-in archive path
to Avro lets users read and infer schema from Avro files packed in a tar
archive without unpacking them first, with the same directory-read parity the
rest of the series guarantees.
### Does this PR introduce _any_ user-facing change?
Yes. When `spark.sql.files.archive.reader.enabled` is set (default
`false`), the Avro data source can now read and infer schema from
`.tar`/`.tar.gz`/`.tgz` archives. With the flag at its default, behavior is
unchanged.
### How was this patch tested?
Added `AvroTarArchiveReadSuite`, which runs the shared archive
read/inference tests from `ArchiveReadSuiteBase` (bound to Avro via
`AvroArchiveReadBase`) over tar containers via `TarArchiveReadBase`, plus the
two Avro-specific streaming-reader regression tests.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #56709 from akshatshenoi-db/archive-avro.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../datasources/AvroArchiveReadBase.scala | 80 ++++++++++++++++++
.../datasources/AvroTarArchiveReadSuite.scala | 28 +++++++
.../org/apache/spark/sql/avro/AvroFileFormat.scala | 92 ++++++++++++++++++++-
.../org/apache/spark/sql/avro/AvroUtils.scala | 94 ++++++++++++++++++++--
.../sql/execution/datasources/ArchiveReader.scala | 31 +++++--
.../datasources/ArchiveReadSuiteBase.scala | 31 +++++--
.../execution/datasources/CSVArchiveReadBase.scala | 21 +----
.../datasources/JSONArchiveReadBase.scala | 21 +----
.../execution/datasources/XMLArchiveReadBase.scala | 20 -----
9 files changed, 336 insertions(+), 82 deletions(-)
diff --git
a/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala
b/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala
new file mode 100644
index 000000000000..0513cfe7db96
--- /dev/null
+++
b/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroArchiveReadBase.scala
@@ -0,0 +1,80 @@
+/*
+ * 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 java.io.{ByteArrayInputStream, IOException}
+
+import org.apache.avro.file.{DataFileConstants, DataFileStream}
+import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
+
+/**
+ * Binds [[ArchiveReadSuiteBase]]'s hooks to Avro, adding the streaming-reader
regression tests
+ * that have no format-agnostic analogue.
+ */
+trait AvroArchiveReadBase extends ArchiveReadSuiteBase {
+
+ override protected def format: String = "avro"
+
+ override protected def fileExtension: String = "avro"
+
+ // Avro is self-describing; no read options needed.
+ override protected def readOptions: Map[String, String] = Map.empty
+
+ override protected def readSchema: String = "id INT, name STRING"
+
+ // Avro takes a single file's writer schema for the whole dataset rather
than unioning fields
+ // across entries, so it opts out of the shared schema-merge tests.
+ override protected def supportsSchemaMerge: Boolean = false
+
+ // Avro's header types are fixed; it does not widen types across entries
like content inference.
+ override protected def excluded: Seq[String] = super.excluded ++ Seq(
+ "archive inference widens a column's type across entries like a directory")
+
+ // ----- Avro-specific tests
-------------------------------------------------
+
+ test("Avro: a truncated entry fails fast instead of spinning") {
+ // A DataFileStream must throw on a truncated entry rather than loop
forever. Cut at the header
+ // and mid-file; the per-test timeout catches a regression to a spin.
+ val full = encodeFile(sampleDf((1, "Alice"), (2, "Bob")))
+ Seq(0, 2, DataFileConstants.MAGIC.length, full.length / 2).foreach { len =>
+ val ex = intercept[Exception] {
+ val stream = new DataFileStream[GenericRecord](
+ new ByteArrayInputStream(full.take(len)), new
GenericDatumReader[GenericRecord]())
+ try while (stream.hasNext) stream.next() finally stream.close()
+ }
+ assert(ex.isInstanceOf[IOException] || ex.isInstanceOf[RuntimeException],
+ s"truncated entry of length $len should fail fast, got $ex")
+ }
+ }
+
+ test("Avro: a drip-fed entry reads fully under partial reads") {
+ // Partial reads (count < requested) must make progress rather than spin.
One byte per read
+ // still reads every record.
+ val full = encodeFile(sampleDf((1, "Alice"), (2, "Bob"), (3, "Carol")))
+ val dripFed = new ByteArrayInputStream(full) {
+ override def read(b: Array[Byte], off: Int, len: Int): Int =
+ super.read(b, off, math.min(1, len))
+ }
+ val stream = new DataFileStream[GenericRecord](dripFed, new
GenericDatumReader[GenericRecord]())
+ try {
+ var count = 0
+ while (stream.hasNext) { stream.next(); count += 1 }
+ assert(count == 3, s"expected all 3 records read through 1-byte reads,
got $count")
+ } finally stream.close()
+ }
+}
diff --git
a/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroTarArchiveReadSuite.scala
b/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroTarArchiveReadSuite.scala
new file mode 100644
index 000000000000..00aa8396ed95
--- /dev/null
+++
b/connector/avro/src/test/scala/org/apache/spark/sql/execution/datasources/AvroTarArchiveReadSuite.scala
@@ -0,0 +1,28 @@
+/*
+ * 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
+
+/**
+ * Reads of Avro files packed in tar archives (`.tar`/`.tar.gz`/`.tgz`): the
shared archive tests
+ * from [[ArchiveReadSuiteBase]] plus the Avro-specific ones from
[[AvroArchiveReadBase]], run over
+ * tar containers via [[TarArchiveReadBase]].
+ */
+class AvroTarArchiveReadSuite
+ extends ArchiveReadSuiteBase
+ with AvroArchiveReadBase
+ with TarArchiveReadBase
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
index 90781d4ad707..aa9992ba6656 100755
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroFileFormat.scala
@@ -22,7 +22,7 @@ import java.io._
import scala.util.control.NonFatal
import org.apache.avro.{LogicalType, LogicalTypes, Schema}
-import org.apache.avro.file.DataFileReader
+import org.apache.avro.file.{DataFileReader, DataFileStream}
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.mapred.FsInput
import org.apache.hadoop.conf.Configuration
@@ -33,7 +33,7 @@ import org.apache.spark.TaskContext
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.{InternalRow, NoopFilters, OrderedFilters}
-import org.apache.spark.sql.execution.datasources.{DataSourceUtils,
FileFormat, OutputWriterFactory, PartitionedFile}
+import org.apache.spark.sql.execution.datasources.{ArchiveReader,
DataSourceUtils, FileFormat, OutputWriterFactory, PartitionedFile}
import org.apache.spark.sql.internal.{SessionStateHelper, SQLConf}
import org.apache.spark.sql.sources.{DataSourceRegister, Filter}
import org.apache.spark.sql.types._
@@ -69,7 +69,14 @@ private[sql] class AvroFileFormat extends FileFormat
override def isSplitable(
sparkSession: SparkSession,
options: Map[String, String],
- path: Path): Boolean = true
+ path: Path): Boolean = {
+ val parsedOptions = new AvroOptions(options,
sparkSession.sessionState.newHadoopConf())
+ if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(path)) {
+ // A tar archive is read as one sequential stream (entry by entry), so
it is never split.
+ return false
+ }
+ true
+ }
override def prepareWrite(
spark: SparkSession,
@@ -95,6 +102,12 @@ private[sql] class AvroFileFormat extends FileFormat
(file: PartitionedFile) => {
val conf = broadcastedConf.value.value
+ if (parsedOptions.archiveFormatEnabled &&
ArchiveReader.isArchivePath(file.toPath)) {
+ // A tar archive (always a single split, see `isSplitable`) is
streamed entry by entry when
+ // archive reads are enabled; otherwise the file is read directly. The
V2 data source has
+ // no archive support, so this dispatch lives here.
+ readArchive(file, conf, parsedOptions, requiredSchema, filters)
+ } else {
val userProvidedSchema = parsedOptions.schema
// TODO Removes this check once `FileFormat` gets a general file
filtering interface method.
@@ -157,6 +170,79 @@ private[sql] class AvroFileFormat extends FileFormat
} else {
Iterator.empty
}
+ }
+ }
+ }
+
+ /**
+ * Streams a tar archive (`.tar`/`.tar.gz`/`.tgz`) entry by entry,
deserializing each entry like a
+ * standalone Avro file via a forward-only [[DataFileStream]] (so the
archive is never unpacked to
+ * disk and memory stays bounded). The whole archive is a single split (see
`isSplitable`). A
+ * fresh datum reader and deserializer are built per entry, since each entry
carries its own
+ * writer schema in its header.
+ *
+ * Kept separate from the per-file reader (rather than dispatched inside it)
because only this V1
+ * read path supports archives; the V2 data source is intentionally left
untouched.
+ */
+ private def readArchive(
+ file: PartitionedFile,
+ conf: Configuration,
+ parsedOptions: AvroOptions,
+ requiredSchema: StructType,
+ filters: Seq[Filter]): Iterator[InternalRow] = {
+ val userProvidedSchema = parsedOptions.schema
+ // Filters depend only on (filters, requiredSchema), which are constant
across entries, so build
+ // them once. The datum reader, writer schema, and rebase mode are
per-entry (each entry carries
+ // its own header schema).
+ val avroFilters = if (SQLConf.get.avroFilterPushDown) {
+ new OrderedFilters(filters, requiredSchema)
+ } else {
+ new NoopFilters
+ }
+ ArchiveReader(file.toPath).readEntries(conf) { (_, in) =>
+ val datumReader = userProvidedSchema match {
+ case Some(schema) => new GenericDatumReader[GenericRecord](schema)
+ case None => new GenericDatumReader[GenericRecord]()
+ }
+ val stream = new DataFileStream[GenericRecord](in, datumReader)
+ val avroSchema = userProvidedSchema.getOrElse(stream.getSchema)
+ val datetimeRebaseMode = DataSourceUtils.datetimeRebaseSpec(
+ stream.getMetaString, parsedOptions.datetimeRebaseModeInRead)
+ val deserializer = new AvroDeserializer(
+ avroSchema,
+ requiredSchema,
+ parsedOptions.positionalFieldMatching,
+ datetimeRebaseMode,
+ avroFilters,
+ parsedOptions.useStableIdForUnionType,
+ parsedOptions.stableIdPrefixForUnionType,
+ parsedOptions.recursiveFieldMaxDepth)
+ // The record is deserialized eagerly in `hasNext` because
`AvroDeserializer#deserialize` may
+ // filter rows (returning None); the stream is closed once its records
are exhausted.
+ new Iterator[InternalRow] with Closeable {
+ private var nextRow: Option[InternalRow] = None
+
+ private def advance(): Unit = {
+ while (nextRow.isEmpty && stream.hasNext) {
+ nextRow =
deserializer.deserialize(stream.next()).asInstanceOf[Option[InternalRow]]
+ }
+ if (nextRow.isEmpty) close()
+ }
+
+ override def hasNext: Boolean = {
+ advance()
+ nextRow.isDefined
+ }
+
+ override def next(): InternalRow = {
+ advance()
+ val row = nextRow.getOrElse(throw new NoSuchElementException("next
on empty iterator"))
+ nextRow = None
+ row
+ }
+
+ override def close(): Unit = stream.close()
+ }
}
}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
index 6f410337b796..b8d2c30b0838 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/avro/AvroUtils.scala
@@ -16,18 +16,18 @@
*/
package org.apache.spark.sql.avro
-import java.io.{FileNotFoundException, IOException}
+import java.io.{Closeable, FileNotFoundException, IOException}
import java.util.Locale
import scala.jdk.CollectionConverters._
import org.apache.avro.{Schema, SchemaFormatter, SchemaParseException}
-import org.apache.avro.file.{DataFileReader, FileReader}
+import org.apache.avro.file.{DataFileReader, DataFileStream, FileReader}
import org.apache.avro.generic.{GenericDatumReader, GenericRecord}
import org.apache.avro.mapred.{AvroOutputFormat, FsInput}
import org.apache.avro.mapreduce.AvroJob
import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.FileStatus
+import org.apache.hadoop.fs.{FileStatus, Path}
import org.apache.hadoop.mapreduce.Job
import org.apache.spark.{SparkException, SparkIllegalArgumentException}
@@ -38,7 +38,7 @@ import org.apache.spark.sql.avro.AvroCompressionCodec._
import org.apache.spark.sql.avro.AvroOptions.IGNORE_EXTENSION
import org.apache.spark.sql.catalyst.{FileSourceOptions, InternalRow}
import org.apache.spark.sql.catalyst.util.CaseInsensitiveMap
-import org.apache.spark.sql.execution.datasources.{DataSourceUtils,
OutputWriterFactory}
+import org.apache.spark.sql.execution.datasources.{ArchiveReader,
DataSourceUtils, OutputWriterFactory}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.util.Utils
@@ -86,8 +86,22 @@ private[sql] object AvroUtils extends Logging {
// User can specify an optional avro json schema.
val avroSchema = parsedOptions.schema
.getOrElse {
- inferAvroSchemaFromFiles(files, conf, parsedOptions.ignoreExtension,
- new
FileSourceOptions(CaseInsensitiveMap(options)).ignoreCorruptFiles)
+ val fileSourceOptions = new
FileSourceOptions(CaseInsensitiveMap(options))
+ // Tar archives are inferred by reading each entry's writer schema
from its Avro header
+ // (streamed, never unpacked to disk), then any loose files. Avro has
no DSv2 reader and
+ // does not merge schemas, so this is V1-only and uses the first
readable writer schema.
+ val (archives, nonArchives) = if
(fileSourceOptions.archiveFormatEnabled) {
+ files.partition(f => ArchiveReader.isArchivePath(f.getPath))
+ } else {
+ (Seq.empty[FileStatus], files)
+ }
+ if (archives.nonEmpty) {
+ inferAvroSchemaFromArchives(archives, nonArchives, conf,
parsedOptions.ignoreExtension,
+ fileSourceOptions.ignoreCorruptFiles,
fileSourceOptions.ignoreMissingFiles)
+ } else {
+ inferAvroSchemaFromFiles(files, conf, parsedOptions.ignoreExtension,
+ fileSourceOptions.ignoreCorruptFiles)
+ }
}
SchemaConverters.toSqlType(
@@ -225,6 +239,74 @@ private[sql] object AvroUtils extends Logging {
}
}
+ /**
+ * Infers an Avro schema from tar archives (`.tar`/`.tar.gz`/`.tgz`) by
reading the writer schema
+ * from the first readable archive entry's Avro header via a forward-only
[[DataFileStream]] --
+ * the archive is streamed, never unpacked to disk, and records are never
scanned. Mirrors
+ * [[inferAvroSchemaFromFiles]]: schema evolution is not supported, so the
first readable writer
+ * schema (across the archives, then any loose files) is used for the whole
dataset, and archives
+ * past the first readable one are never opened.
+ */
+ private def inferAvroSchemaFromArchives(
+ archives: Seq[FileStatus],
+ nonArchives: Seq[FileStatus],
+ conf: Configuration,
+ ignoreExtension: Boolean,
+ ignoreCorruptFiles: Boolean,
+ ignoreMissingFiles: Boolean): Schema = {
+ archives.iterator
+ .flatMap { f =>
+ firstArchiveEntrySchema(f.getPath, conf, ignoreCorruptFiles,
ignoreMissingFiles)
+ }
+ .nextOption()
+ .getOrElse {
+ // No readable schema in any archive; fall back to the loose files.
With none readable
+ // there either, `inferAvroSchemaFromFiles` raises the standard "no
Avro files found" error.
+ inferAvroSchemaFromFiles(nonArchives, conf, ignoreExtension,
ignoreCorruptFiles)
+ }
+ }
+
+ /**
+ * Reads the Avro writer schema of `path`'s first readable entry from its
[[DataFileStream]]
+ * header (records are never scanned), then closes the archive without
reading further entries.
+ * Returns `None` for an empty archive, or for a missing/corrupt archive
under the respective
+ * ignore flag. Because only the first entry is read, a corrupt later entry
never affects
+ * inference -- matching [[inferAvroSchemaFromFiles]], which stops at the
first readable file.
+ */
+ private def firstArchiveEntrySchema(
+ path: Path,
+ conf: Configuration,
+ ignoreCorruptFiles: Boolean,
+ ignoreMissingFiles: Boolean): Option[Schema] = {
+ try {
+ // `readEntries` returns a Closeable iterator; take the first entry's
schema and close it so
+ // the archive stream is released without draining the remaining entries.
+ val entries = ArchiveReader(path).readEntries(conf) { (_, in) =>
+ val stream = new DataFileStream[GenericRecord](in, new
GenericDatumReader[GenericRecord]())
+ try {
+ Iterator.single(stream.getSchema)
+ } finally {
+ stream.close()
+ }
+ }
+ try {
+ if (entries.hasNext) Some(entries.next()) else None
+ } finally {
+ entries match {
+ case c: Closeable => c.close()
+ case _ =>
+ }
+ }
+ } catch {
+ case _: FileNotFoundException if ignoreMissingFiles =>
+ logWarning(log"Skipped missing archive: ${MDC(PATH, path)}")
+ None
+ case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted archive: ${MDC(PATH, path)}", e)
+ None
+ }
+ }
+
// The trait provides iterator-like interface for reading records from an
Avro file,
// deserializing and returning them as internal rows.
trait RowReader {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
index 84daeb92fe8e..75f04f8f38c6 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ArchiveReader.scala
@@ -170,8 +170,16 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
/** Opens the archive as a tar stream, transparently decompressing `.tar.gz`
/ `.tgz`. */
private def openTarStream(conf: Configuration): TarArchiveInputStream = {
val base = CodecStreams.createInputStreamWithCloseResource(conf, path)
- val tarBytes = if (needsExplicitGunzip) new GZIPInputStream(base) else base
- new TarArchiveInputStream(tarBytes)
+ try {
+ // GZIPInputStream reads the gzip header in its constructor, so a
corrupt archive can throw
+ // here -- after `base` is already open -- and `base` must not leak.
+ val tarBytes = if (needsExplicitGunzip) new GZIPInputStream(base) else
base
+ new TarArchiveInputStream(tarBytes)
+ } catch {
+ case NonFatal(e) =>
+ try base.close() catch { case NonFatal(_) => }
+ throw e
+ }
}
/**
@@ -199,7 +207,7 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ =>
cleanup()))
- new Iterator[T] with Closeable {
+ val entries = new Iterator[T] with Closeable {
private var currentIter: Iterator[T] = Iterator.empty
private var done = false
@@ -228,9 +236,6 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
}
}
- // Open the first entry eagerly so construction reflects the archive's
first entry.
- advance()
-
override def hasNext: Boolean = {
advance()
!done && currentIter.hasNext
@@ -247,5 +252,19 @@ class TarArchiveReader(path: Path) extends
ArchiveReader(path) {
cleanup()
}
}
+
+ // Open the first entry eagerly so the construction cost (and any failure)
surfaces here rather
+ // than at the first `hasNext`. A corrupt archive throws before the caller
ever holds the
+ // iterator, leaving it nothing to close: executors release the stream
through the task-
+ // completion listener, but driver-side callers (e.g. Avro's header-only
schema inference) have
+ // no task, so close it here before propagating.
+ try {
+ entries.hasNext
+ } catch {
+ case NonFatal(e) =>
+ cleanup()
+ throw e
+ }
+ entries
}
}
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 b9bec4fd756a..8ec73fbfefda 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
@@ -64,8 +64,26 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
/** Schema used to read the sample data produced by [[sampleDf]]. */
protected def readSchema: String
- /** Encodes `df` as the bytes of a single data file of [[format]], honoring
`writeOptions`. */
- protected def encodeFile(df: DataFrame, writeOptions: Map[String, String]):
Array[Byte]
+ /**
+ * Encodes `df` as the bytes of a single data file of [[format]], honoring
`writeOptions` (plus
+ * [[readOptions]], so e.g. CSV writes with the same `header` mode it
reads). Coalesces to one
+ * partition and returns that single part file's bytes.
+ */
+ protected def encodeFile(df: DataFrame, writeOptions: Map[String, String]):
Array[Byte] = {
+ val dir = Utils.createTempDir(namePrefix = "archive-test-encode")
+ try {
+ df.coalesce(1).write.format(format)
+ .options(readOptions ++ writeOptions)
+ .mode("overwrite").save(dir.getCanonicalPath)
+ val parts = dir.listFiles().filter { f =>
+ f.isFile && !f.getName.startsWith("_") && !f.getName.startsWith(".") &&
+ !f.getName.endsWith(".crc")
+ }
+ assert(parts.length == 1,
+ s"expected exactly one data file, got: ${parts.map(_.getName).toList}")
+ Files.readAllBytes(parts.head.toPath)
+ } finally Utils.deleteRecursively(dir)
+ }
/** Encodes `df` as a single data file using only the format's default write
options. */
protected final def encodeFile(df: DataFrame): Array[Byte] = encodeFile(df,
Map.empty)
@@ -107,11 +125,10 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
spark.read.format(format).options(readOptions ++
extraOptions).schema(schema).load(path)
/**
- * Whether this format infers its read schema from the textual content of
the files (CSV, JSON,
- * XML), as opposed to carrying an embedded schema (Avro, Parquet) or none
(text). Gates the
- * shared schema-inference tests. A format that needs an option to trigger
inference (e.g. CSV's
- * `inferSchema`) also overrides [[inferenceOptions]]; a format that does
not infer overrides this
- * to false.
+ * Whether this format can infer a read schema from the files (content or an
embedded header).
+ * Gates the shared schema-inference tests; a format that cannot infer (e.g.
ORC) sets it false.
+ * A format that infers but does not widen types across entries excludes
that one test via
+ * [[excluded]] (see Avro). CSV-style inference triggers also override
[[inferenceOptions]].
*/
protected def supportsSchemaInference: Boolean = true
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
index a18d950fb0f0..9ad49235b3fb 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/CSVArchiveReadBase.scala
@@ -21,10 +21,9 @@ import java.io.File
import java.nio.charset.StandardCharsets
import java.nio.file.Files
-import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.StringType
-import org.apache.spark.util.Utils
/**
* Binds [[ArchiveReadSuiteBase]]'s file-format hooks to CSV. The
header-mode-specific tests live in
@@ -53,24 +52,6 @@ trait CSVArchiveReadBase extends ArchiveReadSuiteBase {
override protected def supportsSchemaMerge: Boolean = false
- override protected def encodeFile(
- df: DataFrame,
- writeOptions: Map[String, String]): Array[Byte] = {
- val dir = Utils.createTempDir(namePrefix = "archive-test-encode")
- try {
- df.coalesce(1).write.format("csv")
- .options(Map("header" -> header.toString) ++ writeOptions)
- .mode("overwrite").save(dir.getCanonicalPath)
- val parts = dir.listFiles().filter { f =>
- f.isFile && !f.getName.startsWith("_") && !f.getName.startsWith(".") &&
- !f.getName.endsWith(".crc")
- }
- assert(parts.length == 1,
- s"expected exactly one data file, got: ${parts.map(_.getName).toList}")
- Files.readAllBytes(parts.head.toPath)
- } finally Utils.deleteRecursively(dir)
- }
-
/** Raw CSV bytes, for tests that need precise control over the row layout.
*/
protected def csvBytes(s: String): Array[Byte] =
s.getBytes(StandardCharsets.UTF_8)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
index fdad5df4b12d..51e89832a4e5 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/JSONArchiveReadBase.scala
@@ -21,10 +21,9 @@ import java.io.File
import java.nio.charset.StandardCharsets
import java.nio.file.Files
-import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.AnalysisException
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{NullType, StringType}
-import org.apache.spark.util.Utils
/**
* Binds [[ArchiveReadSuiteBase]]'s file-format hooks to JSON. JSON opts into
the shared
@@ -44,24 +43,6 @@ trait JSONArchiveReadBase extends ArchiveReadSuiteBase {
override protected def readSchema: String = "id INT, name STRING"
- override protected def encodeFile(
- df: DataFrame,
- writeOptions: Map[String, String]): Array[Byte] = {
- val dir = Utils.createTempDir(namePrefix = "archive-test-encode")
- try {
- df.coalesce(1).write.format("json")
- .options(writeOptions)
- .mode("overwrite").save(dir.getCanonicalPath)
- val parts = dir.listFiles().filter { f =>
- f.isFile && !f.getName.startsWith("_") && !f.getName.startsWith(".") &&
- !f.getName.endsWith(".crc")
- }
- assert(parts.length == 1,
- s"expected exactly one data file, got: ${parts.map(_.getName).toList}")
- Files.readAllBytes(parts.head.toPath)
- } finally Utils.deleteRecursively(dir)
- }
-
/** Raw JSON bytes, for tests that need precise control over the record
layout. */
protected def jsonBytes(s: String): Array[Byte] =
s.getBytes(StandardCharsets.UTF_8)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
index ef6a27a99e25..71f97dd32637 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/XMLArchiveReadBase.scala
@@ -21,9 +21,7 @@ import java.io.File
import java.nio.charset.StandardCharsets
import java.nio.file.Files
-import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types.StringType
-import org.apache.spark.util.Utils
/**
* Binds [[ArchiveReadSuiteBase]]'s file-format hooks to XML. XML opts into
the shared
@@ -50,24 +48,6 @@ trait XMLArchiveReadBase extends ArchiveReadSuiteBase {
// complex types) and runs the full shared test set. Inference needs no
trigger option, so
// `inferenceOptions` keeps its empty default.
- override protected def encodeFile(
- df: DataFrame,
- writeOptions: Map[String, String]): Array[Byte] = {
- val dir = Utils.createTempDir(namePrefix = "archive-test-encode")
- try {
- df.coalesce(1).write.format("xml")
- .options(Map("rowTag" -> rowTag) ++ writeOptions)
- .mode("overwrite").save(dir.getCanonicalPath)
- val parts = dir.listFiles().filter { f =>
- f.isFile && !f.getName.startsWith("_") && !f.getName.startsWith(".") &&
- !f.getName.endsWith(".crc")
- }
- assert(parts.length == 1,
- s"expected exactly one data file, got: ${parts.map(_.getName).toList}")
- Files.readAllBytes(parts.head.toPath)
- } finally Utils.deleteRecursively(dir)
- }
-
/** Raw XML bytes, for tests that need precise control over the record
layout. */
protected def xmlBytes(s: String): Array[Byte] =
s.getBytes(StandardCharsets.UTF_8)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
