This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 82dba8de97ba [SPARK-57321][SQL] Infer CSV schema from tar archives
82dba8de97ba is described below
commit 82dba8de97ba592b949d5bf863c47a1a9c7a333d
Author: akshatshenoi-db <[email protected]>
AuthorDate: Thu Jun 11 10:53:42 2026 -0700
[SPARK-57321][SQL] Infer CSV schema from tar archives
### What changes were proposed in this pull request?
> **Stacked on #56193** (CSV tar archive read support). Please review/merge
that PR first; until it merges, this PR's diff will also include its commits.
The inference-specific changes are in the top commit.
Adds CSV **schema inference** for tar archives (`.tar`/`.tar.gz`/`.tgz`),
building on the archive read support in #56193. When
`spark.sql.files.archive.reader.enabled` is set and an input path is a tar
archive, `CSVDataSource.inferSchema` infers it by streaming the archive's
entries through the existing `ArchiveReader` (never unpacked to disk).
Inference runs as a **single `CSVInferSchema` pass over all inputs**: every
archive entry and every loose CSV file is tokenized into one `tokenRDD`, keyed
on the first input's header, sampled, and inferred in one pass (the column
count is fixed by the first header, and `NullType` columns survive until the
final `toStructFields`). Tokenization is **mode-specific**, matching how the
scan reads each input: in the default mode each entry is tokenized line-by-line
(so a quoted field contai [...]
Archive scanning is wired into the V1 file source only (#56193), so archive
schema inference is gated on whether the calling scan path can read archives:
`CSVFileFormat` (V1) passes `supportsArchiveScan = true`, while the DSv2
`CSVTable` passes `false`. When the DSv2 path encounters an archive input,
inference returns `None` so the read fails loudly with
`UNABLE_TO_INFER_SCHEMA`, rather than letting the V2 scan parse raw archive
bytes as CSV.
The enablement flag is read from `FileSourceOptions.archiveFormatEnabled`
(added in #56193); no new config is introduced. The config doc is updated to
note archives are supported during both scan and schema inference.
### Why are the changes needed?
The archive feature was split into two PRs to keep each reviewable: #56193
adds reading, and this PR adds schema inference so that `inferSchema=true` (and
the default inference when no schema is supplied) works for archives the same
way it does for a directory of CSV files. Without it, reading an archive
without an explicit schema errors with `UNABLE_TO_INFER_SCHEMA`.
### Does this PR introduce _any_ user-facing change?
No. The capability is behind the `spark.sql.files.archive.reader.enabled`
config (default `false`, introduced in #56193); this PR only extends that
opt-in feature to schema inference.
### How was this patch tested?
Added inference tests to `CSVArchiveReadBase`, run in both header and
headerless modes by `CSVHeaderTarArchiveReadSuite` /
`CSVHeaderlessTarArchiveReadSuite`:
- an archive infers the same schema as a directory of the same files;
- all archive formats (`.tar`/`.tar.gz`/`.tgz`) infer the same schema;
- a corrupt archive among good ones is skipped under `ignoreCorruptFiles`;
- a column's type is widened across entries (e.g. an integer column in one
entry and a string column in another infer as string);
- archive entries and loose files in the same directory infer a single
merged schema;
- a column that is empty in an archived file but typed in a loose file is
**not** collapsed to `StringType` — it widens with the loose value, matching a
single-pass directory read;
- when entries have different widths, the column count is fixed by the
first entry's header, and the columns keep their inferred types (pinning the
per-entry header drop);
- inference uses the same record model as the line-based scan: a quoted
field with an embedded newline is split across rows in the default mode, so the
archive infers the same schema as that entry read as a loose file;
- the same parity holds under `multiLine`, where the quoted newline stays a
single record — pinning the whole-stream tokenization branch;
- on the DSv2 path (`CSVTable`) an archive input keeps raising
`UNABLE_TO_INFER_SCHEMA` (inference returns `None`) instead of inferring a
schema the V2 scan would mis-read.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #56254 from akshatshenoi-db/archive-format-schema-inference.
Authored-by: akshatshenoi-db <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 89dff6beac6e092ea485b5f40a18bee916a89da6)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 4 +-
.../execution/datasources/csv/CSVDataSource.scala | 164 ++++++++++++++----
.../execution/datasources/csv/CSVFileFormat.scala | 5 +-
.../execution/datasources/v2/csv/CSVTable.scala | 6 +-
.../datasources/ArchiveReadSuiteBase.scala | 6 +-
.../execution/datasources/CSVArchiveReadBase.scala | 185 ++++++++++++++++++++-
6 files changed, 321 insertions(+), 49 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 80a55cf5802d..971eddf7bef9 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -2725,8 +2725,8 @@ object SQLConf {
val ARCHIVE_FORMAT_READER_ENABLED =
buildConf("spark.sql.files.archive.reader.enabled")
.doc("When true, the CSV data source can read tar archives (.tar, .tar.gz,
.tgz): each " +
"archive is read as a single split and its entries are streamed through
the CSV parser " +
- "(never unpacked to disk), as if the entries were separate CSV files.
Only the CSV data " +
- "source supports reading archives.")
+ "(never unpacked to disk), as if the entries were separate CSV files,
both during scan " +
+ "and schema inference. Only the CSV data source supports reading
archives.")
.version("5.0.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
index 76a34630a4d5..5170ce8a3da3 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
@@ -48,7 +48,7 @@ import org.apache.spark.util.Utils
/**
* Common functions for parsing CSV files
*/
-abstract class CSVDataSource extends Serializable {
+abstract class CSVDataSource extends Serializable with Logging {
def isSplitable: Boolean
/**
@@ -67,16 +67,22 @@ abstract class CSVDataSource extends Serializable {
final def inferSchema(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
- parsedOptions: CSVOptions): Option[StructType] = {
+ parsedOptions: CSVOptions,
+ supportsArchiveScan: Boolean): Option[StructType] = {
parsedOptions.singleVariantColumn match {
case Some(columnName) => Some(StructType(Array(StructField(columnName,
VariantType))))
case None =>
- if (parsedOptions.archiveFormatEnabled &&
- inputPaths.exists(f => ArchiveReader.isArchivePath(f.getPath))) {
- // Schema inference is not yet supported for tar archives. Returning
None makes Spark
- // raise its standard "Unable to infer schema ... It must be
specified manually" error
- // (UNABLE_TO_INFER_SCHEMA), so reading an archive requires an
explicit `.schema(...)`.
- // Inferring a schema by streaming archive entries is planned as a
follow-up.
+ val hasArchive = parsedOptions.archiveFormatEnabled &&
+ inputPaths.exists(f => ArchiveReader.isArchivePath(f.getPath))
+ if (hasArchive && supportsArchiveScan) {
+ // Archives (and any loose files alongside them) are inferred in a
single CSVInferSchema
+ // pass over all inputs -- archive entries are streamed, never
unpacked -- so the result
+ // matches what the scan returns for the same files.
+ Some(inferWithArchives(sparkSession, inputPaths, parsedOptions))
+ } else if (hasArchive) {
+ // The caller's scan path cannot read archives (e.g. the DSv2
reader), so refuse to infer
+ // a schema when any input is an archive: returning None raises
UNABLE_TO_INFER_SCHEMA,
+ // which fails loudly instead of letting the scan parse raw archive
bytes as CSV.
None
} else if (inputPaths.nonEmpty) {
Some(infer(sparkSession, inputPaths, parsedOptions))
@@ -91,6 +97,19 @@ abstract class CSVDataSource extends Serializable {
inputPaths: Seq[FileStatus],
parsedOptions: CSVOptions): StructType
+ /**
+ * Tokenizes one input's bytes -- a whole loose file or a single
decompressed archive entry --
+ * into CSV rows the same way this mode's scan reads it: line by line for
+ * [[TextInputCSVDataSource]] (so a quoted field with an embedded newline
splits across rows
+ * exactly as the read does) and as one continuous stream for
[[MultiLineCSVDataSource]]. This
+ * input's own first row (its header) is dropped when `dropHeader` is set.
Used only by
+ * [[inferWithArchives]]; the stream is not closed here.
+ */
+ protected def tokenizeForInference(
+ in: InputStream,
+ dropHeader: Boolean,
+ options: CSVOptions): Iterator[Array[String]]
+
/**
* Streams a tar archive (`.tar`/`.tar.gz`/`.tgz`) entry by entry through
the CSV parser without
* unpacking it to disk. The whole archive is a single split (see
`CSVFileFormat.isSplitable`); a
@@ -130,6 +149,60 @@ abstract class CSVDataSource extends Serializable {
parseEntry(parser, headerChecker, in)
}
}
+
+ /**
+ * Infers a CSV schema when at least one input is a tar archive. Every
archive entry is streamed
+ * (never unpacked to disk) and every loose file is read, each tokenized the
same way this mode's
+ * scan reads it (see [[tokenizeForInference]]), and all of them feed a
single [[CSVInferSchema]]
+ * pass keyed on the first input's header. The column count is fixed by the
first header and
+ * `NullType` columns survive to the final `toStructFields`, so the inferred
schema matches what
+ * the scan returns for the same inputs. A corrupt/missing input is skipped
as a unit (a whole
+ * archive or a whole file) when `ignoreCorruptFiles`/`ignoreMissingFiles`
are set. Both the
+ * header and sampling passes honor those flags, so archive inference is
strictly more forgiving
+ * here than [[MultiLineCSVDataSource.infer]], whose sampling pass has no
such guard.
+ */
+ private def inferWithArchives(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ parsedOptions: CSVOptions): StructType = {
+ val baseRdd = CSVDataSource.createBaseRdd(sparkSession, inputPaths,
parsedOptions)
+ def tokens(dropHeader: Boolean): RDD[Array[String]] = baseRdd.flatMap {
stream =>
+ val path = new Path(stream.getPath())
+ try {
+ if (ArchiveReader.isArchivePath(path)) {
+ ArchiveReader(path).readEntries(stream.getConfiguration) { (_, in) =>
+ tokenizeForInference(in, dropHeader, parsedOptions)
+ }
+ } else {
+ tokenizeForInference(
+
CodecStreams.createInputStreamWithCloseResource(stream.getConfiguration, path),
+ dropHeader, parsedOptions)
+ }
+ } catch {
+ case e: FileNotFoundException if parsedOptions.ignoreMissingFiles =>
+ logWarning(log"Skipped missing input: ${MDC(PATH,
stream.getPath())}", e)
+ Iterator.empty
+ case e: FileNotFoundException => throw e
+ case e @ (_: RuntimeException | _: IOException) if
parsedOptions.ignoreCorruptFiles =>
+ logWarning(log"Skipped the corrupted input: ${MDC(PATH,
stream.getPath())}", e)
+ Iterator.empty
+ case NonFatal(e) =>
+ throw QueryExecutionErrors.cannotReadFilesError(
+ e, SparkPath.fromPathString(stream.getPath()).urlEncoded)
+ }
+ }
+ tokens(dropHeader = false).take(1).headOption match {
+ case Some(firstRow) =>
+ val caseSensitive =
sparkSession.sessionState.conf.caseSensitiveAnalysis
+ val header = CSVUtils.makeSafeHeader(firstRow, caseSensitive,
parsedOptions)
+ val sampled = CSVUtils.sample(tokens(dropHeader =
parsedOptions.headerFlag), parsedOptions)
+ SQLExecution.withSQLConfPropagated(sparkSession) {
+ new CSVInferSchema(parsedOptions).infer(sampled, header)
+ }
+ case None =>
+ StructType(Nil)
+ }
+ }
}
object CSVDataSource extends Logging {
@@ -141,6 +214,33 @@ object CSVDataSource extends Logging {
}
}
+ /**
+ * One `PortableDataStream` per input file (the whole file, never split),
shared by the multiLine
+ * schema-inference path and the archive inference path.
+ */
+ private[csv] def createBaseRdd(
+ sparkSession: SparkSession,
+ inputPaths: Seq[FileStatus],
+ options: CSVOptions): RDD[PortableDataStream] = {
+ val paths = inputPaths.map(_.getPath)
+ val name = paths.mkString(",")
+ val job =
Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
+ options.parameters))
+ FileInputFormat.setInputPaths(job, paths: _*)
+ val conf = job.getConfiguration
+
+ val rdd = new BinaryFileRDD(
+ sparkSession.sparkContext,
+ classOf[StreamInputFormat],
+ classOf[String],
+ classOf[PortableDataStream],
+ conf,
+ sparkSession.sparkContext.defaultMinPartitions)
+
+ // Only returns `PortableDataStream`s without paths.
+ rdd.setName(s"CSVFile: $name").values
+ }
+
/**
* Returns a function that sets the header column names used in
singleVariantColumn mode. The
* returned function takes an optional input, which is the header column
names potentially read by
@@ -221,6 +321,19 @@ object TextInputCSVDataSource extends CSVDataSource {
}
}
+ override protected def tokenizeForInference(
+ in: InputStream,
+ dropHeader: Boolean,
+ options: CSVOptions): Iterator[Array[String]] = {
+ // Match the line-based scan (`readArchive` -> `entryLines` ->
`parseIterator`): split into
+ // lines, drop comments/blanks, drop this input's header line, then parse
each line on its own,
+ // so a quoted field with an embedded newline splits across rows exactly
as the read does.
+ val lines = CSVUtils.filterCommentAndEmpty(entryLines(in, options),
options)
+ val rows = if (dropHeader && lines.hasNext) { lines.next(); lines } else
lines
+ val parser = new CsvParser(options.asParserSettings)
+ rows.map(parser.parseLine)
+ }
+
override def infer(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
@@ -286,7 +399,7 @@ object TextInputCSVDataSource extends CSVDataSource {
}
}
-object MultiLineCSVDataSource extends CSVDataSource with Logging {
+object MultiLineCSVDataSource extends CSVDataSource {
override val isSplitable: Boolean = false
override def readFile(
@@ -316,11 +429,19 @@ object MultiLineCSVDataSource extends CSVDataSource with
Logging {
UnivocityParser.parseStream(in, parser, headerChecker, requiredSchema)
}
+ override protected def tokenizeForInference(
+ in: InputStream,
+ dropHeader: Boolean,
+ options: CSVOptions): Iterator[Array[String]] =
+ // The whole input is one continuous stream, exactly as the multi-line
scan reads it.
+ UnivocityParser.tokenizeStream(
+ in, dropHeader, new CsvParser(options.asParserSettings), options.charset)
+
override def infer(
sparkSession: SparkSession,
inputPaths: Seq[FileStatus],
parsedOptions: CSVOptions): StructType = {
- val csv = createBaseRdd(sparkSession, inputPaths, parsedOptions)
+ val csv = CSVDataSource.createBaseRdd(sparkSession, inputPaths,
parsedOptions)
val ignoreCorruptFiles = parsedOptions.ignoreCorruptFiles
val ignoreMissingFiles = parsedOptions.ignoreMissingFiles
csv.flatMap { lines =>
@@ -366,27 +487,4 @@ object MultiLineCSVDataSource extends CSVDataSource with
Logging {
StructType(Nil)
}
}
-
- private def createBaseRdd(
- sparkSession: SparkSession,
- inputPaths: Seq[FileStatus],
- options: CSVOptions): RDD[PortableDataStream] = {
- val paths = inputPaths.map(_.getPath)
- val name = paths.mkString(",")
- val job =
Job.getInstance(sparkSession.sessionState.newHadoopConfWithOptions(
- options.parameters))
- FileInputFormat.setInputPaths(job, paths: _*)
- val conf = job.getConfiguration
-
- val rdd = new BinaryFileRDD(
- sparkSession.sparkContext,
- classOf[StreamInputFormat],
- classOf[String],
- classOf[PortableDataStream],
- conf,
- sparkSession.sparkContext.defaultMinPartitions)
-
- // Only returns `PortableDataStream`s without paths.
- rdd.setName(s"CSVFile: $name").values
- }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
index 80d470e1919e..26ad735e89aa 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVFileFormat.scala
@@ -57,7 +57,10 @@ case class CSVFileFormat() extends TextBasedFileFormat with
DataSourceRegister {
options: Map[String, String],
files: Seq[FileStatus]): Option[StructType] = {
val parsedOptions = getCsvOptions(sparkSession, options)
- CSVDataSource(parsedOptions).inferSchema(sparkSession, files,
parsedOptions)
+ // The v1 file format routes archives to `readArchive` (see
`buildReader`), so archive schema
+ // inference is supported here.
+ CSVDataSource(parsedOptions)
+ .inferSchema(sparkSession, files, parsedOptions, supportsArchiveScan =
true)
}
override def prepareWrite(
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
index 221196c86445..a1d9f2f7c90b 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/csv/CSVTable.scala
@@ -46,7 +46,11 @@ case class CSVTable(
columnPruning = sparkSession.sessionState.conf.csvColumnPruning,
sparkSession.sessionState.conf.sessionLocalTimeZone)
- CSVDataSource(parsedOptions).inferSchema(sparkSession, files,
parsedOptions)
+ // The DSv2 reader does not route archives to `readArchive` (it calls
`readFile` directly), so
+ // archive scans aren't supported here; pass supportsArchiveScan = false
so an archive input
+ // keeps failing with UNABLE_TO_INFER_SCHEMA rather than having its raw
bytes parsed as CSV.
+ CSVDataSource(parsedOptions)
+ .inferSchema(sparkSession, files, parsedOptions, supportsArchiveScan =
false)
}
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
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 9d0300ec4117..903adb9f98bb 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
@@ -90,9 +90,9 @@ trait ArchiveReadSuiteBase extends QueryTest with
SharedSparkSession {
/** Entry file name for the i-th data file in an archive. */
protected def entryName(i: Int): String = s"part-$i.$fileExtension"
- /** Provides an archive-extensioned path inside a fresh temp dir to `f`. */
- protected def withArchiveFile(
- extension: String = archiveExtensions.head)(f: File => Unit): Unit = {
+ /** Provides an archive-extensioned path inside a fresh temp dir to `f` and
returns its result. */
+ protected def withArchiveFile[T](
+ extension: String = archiveExtensions.head)(f: File => T): T = {
val dir = Utils.createTempDir(namePrefix = "archive-test")
try f(new File(dir, s"archive.$extension")) finally
Utils.deleteRecursively(dir)
}
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 93dc8db5750f..e8076cfb9427 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
@@ -17,11 +17,13 @@
package org.apache.spark.sql.execution.datasources
+import java.io.File
import java.nio.charset.StandardCharsets
import java.nio.file.Files
-import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.DataFrame
+import org.apache.spark.sql.{AnalysisException, DataFrame}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.StringType
import org.apache.spark.util.Utils
/**
@@ -63,16 +65,181 @@ trait CSVArchiveReadBase extends ArchiveReadSuiteBase {
/** 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)
- test("CSV: reading an archive without a schema fails (inference not yet
supported)") {
- // Schema inference for archives is a follow-up; until then an explicit
schema is required, and
- // an inference attempt raises Spark's standard UNABLE_TO_INFER_SCHEMA
error.
+ test("CSV: archive infers the same schema as a directory of the same files")
{
+ val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol")))
+ .zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
+ withArchiveFile() { archive =>
+ writeArchive(archive, entries)
+ val archiveSchema =
spark.read.options(readOptions).option("inferSchema", "true")
+ .format(format).load(archive.getCanonicalPath).schema
+ withTempDir { dir =>
+ entries.foreach { case (n, b) => Files.write(new File(dir, n).toPath,
b) }
+ val dirSchema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(dir.getCanonicalPath).schema
+ assert(archiveSchema == dirSchema,
+ s"inference parity broken; archive=$archiveSchema dir=$dirSchema")
+ }
+ }
+ }
+
+ test("CSV: all archive formats infer the same schema") {
+ val entries = Seq(sampleDf((1, "Alice"), (2, "Bob")), sampleDf((3,
"Carol")))
+ .zipWithIndex.map { case (p, i) => entryName(i) -> encodeFile(p) }
+ val schemas = archiveExtensions.map { ext =>
+ withArchiveFile(ext) { archive =>
+ writeArchive(archive, entries)
+ spark.read.options(readOptions).option("inferSchema", "true")
+ .format(format).load(archive.getCanonicalPath).schema
+ }
+ }
+ assert(schemas.distinct.size == 1,
+ s"archive formats inferred different schemas:
${archiveExtensions.zip(schemas)}")
+ }
+
+ /** CSV bytes for `rows`, prefixed with a `cols` header line when [[header]]
is set. */
+ private def csvEntry(cols: String, rows: String*): Array[Byte] =
+ csvBytes((if (header) cols +: rows else rows).mkString("", "\n", "\n"))
+
+ test("CSV: inference skips a corrupt archive among good ones
(ignoreCorruptFiles)") {
+ withTempDir { dir =>
+ val good = sampleDf((1, "Alice"), (2, "Bob"))
+ writeArchive(new File(dir, s"good.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(good)))
+ writeCorruptArchive(new File(dir, s"bad.$corruptArchiveExtension"))
+ withSQLConf(SQLConf.IGNORE_CORRUPT_FILES.key -> "true") {
+ val schema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(dir.getCanonicalPath).schema
+ withTempDir { onlyGood =>
+ Files.write(new File(onlyGood, entryName(0)).toPath,
encodeFile(good))
+ val expected = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(onlyGood.getCanonicalPath).schema
+ assert(schema == expected,
+ s"corrupt archive not skipped during inference; got $schema, want
$expected")
+ }
+ }
+ }
+ }
+
+ test("CSV: inference widens a column's type across archive entries") {
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ entryName(0) -> csvEntry("c", "1", "2"),
+ entryName(1) -> csvEntry("c", "x")))
+ val schema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(archive.getCanonicalPath).schema
+ assert(schema.length == 1 && schema.head.dataType == StringType,
+ s"expected the column widened to string across entries, got $schema")
+ }
+ }
+
+ test("CSV: inference merges archive entries with loose files in the same
directory") {
+ withTempDir { dir =>
+ val inArchive = sampleDf((1, "Alice"), (2, "Bob"))
+ val loose = sampleDf((3, "Carol"))
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
+ Seq(entryName(0) -> encodeFile(inArchive)))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath,
encodeFile(loose))
+ val schema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(dir.getCanonicalPath).schema
+ withTempDir { looseDir =>
+ Files.write(new File(looseDir, entryName(0)).toPath,
encodeFile(inArchive))
+ Files.write(new File(looseDir, s"loose.$fileExtension").toPath,
encodeFile(loose))
+ val expected = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(looseDir.getCanonicalPath).schema
+ assert(schema == expected,
+ s"mixed archive+loose inference diverged from directory; got
$schema, want $expected")
+ }
+ }
+ }
+
+ test("CSV: a column empty in the archive but typed in a loose file is not
collapsed to string") {
+ // One inference pass over all inputs keeps the empty column NullType
until the end, so it
+ // widens with the loose file's Int. Merging two already-finished schemas
would have collapsed
+ // the archive side to String first and yielded String here.
+ withTempDir { dir =>
+ writeArchive(new File(dir, s"data.${archiveExtensions.head}"),
+ Seq(entryName(0) -> csvEntry("a,b", "1,", "2,")))
+ Files.write(new File(dir, s"loose.$fileExtension").toPath,
csvEntry("a,b", "3,4"))
+ val schema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(dir.getCanonicalPath).schema
+ assert(schema.length == 2 && schema(1).dataType != StringType,
+ s"empty-in-archive column should widen with the loose Int, not
collapse to String: $schema")
+ }
+ }
+
+ test("CSV: archive inference fixes the column count from the first entry's
header") {
+ // The first entry has two columns, the second three; one inference pass
keys on the first
+ // header, so the extra column in the later entry is dropped -- the same
first-header-width
+ // model a single-pass directory read uses.
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(
+ entryName(0) -> csvEntry("a,b", "1,2"),
+ entryName(1) -> csvEntry("a,b,c", "3,4,5")))
+ val schema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(archive.getCanonicalPath).schema
+ assert(schema.length == 2 && schema.forall(_.dataType != StringType),
+ s"expected 2 typed columns fixed by the first entry's header, got
$schema")
+ }
+ }
+
+ test("CSV: inference uses the same record model as the scan (quoted embedded
newline)") {
+ // In default (non-multiLine) mode the scan reads line by line, so a
quoted field containing a
+ // newline is split across rows; inference must tokenize the archived
entry the same way, so it
+ // infers the same schema as that entry read as a loose file (rather than
parsing the entry as
+ // one continuous stream and disagreeing with the read).
+ val entry = csvEntry("a,b", "\"x\ny\",2")
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(entryName(0) -> entry))
+ val archiveSchema =
spark.read.options(readOptions).option("inferSchema", "true")
+ .format(format).load(archive.getCanonicalPath).schema
+ withTempDir { dir =>
+ Files.write(new File(dir, entryName(0)).toPath, entry)
+ val dirSchema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .format(format).load(dir.getCanonicalPath).schema
+ assert(archiveSchema == dirSchema,
+ s"archive inference diverged from the line-based read; " +
+ s"archive=$archiveSchema dir=$dirSchema")
+ }
+ }
+ }
+
+ test("CSV: archive inference under multiLine matches the scan (quoted
embedded newline)") {
+ // Under multiLine the scan reads each entry as one stream, so a quoted
embedded newline is one
+ // record; inference dispatches to the same stream model and agrees with
that entry read as a
+ // loose multiLine file. Pins the multiLine branch of the per-mode
tokenization dispatch.
+ val entry = csvEntry("a,b", "\"x\ny\",2")
+ withArchiveFile() { archive =>
+ writeArchive(archive, Seq(entryName(0) -> entry))
+ val archiveSchema =
spark.read.options(readOptions).option("inferSchema", "true")
+ .option("multiLine",
"true").format(format).load(archive.getCanonicalPath).schema
+ withTempDir { dir =>
+ Files.write(new File(dir, entryName(0)).toPath, entry)
+ val dirSchema = spark.read.options(readOptions).option("inferSchema",
"true")
+ .option("multiLine",
"true").format(format).load(dir.getCanonicalPath).schema
+ assert(archiveSchema == dirSchema,
+ s"multiLine archive inference diverged from the multiLine read; " +
+ s"archive=$archiveSchema dir=$dirSchema")
+ }
+ }
+ }
+
+ test("CSV: the DSv2 path refuses to infer a schema for an archive
(UNABLE_TO_INFER_SCHEMA)") {
+ // Archive scanning is wired into the V1 file source only, so the DSv2
reader cannot read
+ // archives. On the V2 path inference must keep returning None for an
archive input -- raising
+ // UNABLE_TO_INFER_SCHEMA -- rather than inferring a schema and letting
the V2 scan parse the
+ // raw archive bytes as CSV. Forcing csv off the V1 source list routes the
read through
+ // CSVTable.
withArchiveFile() { archive =>
writeArchive(archive, Seq(entryName(0) -> encodeFile(sampleDf((1,
"Alice"), (2, "Bob")))))
- val e = intercept[AnalysisException] {
-
spark.read.format(format).options(readOptions).load(archive.getCanonicalPath)
+ withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "") {
+ val e = intercept[AnalysisException] {
+ spark.read.options(readOptions).option("inferSchema", "true")
+ .format(format).load(archive.getCanonicalPath)
+ }
+ assert(e.getCondition == "UNABLE_TO_INFER_SCHEMA",
+ s"expected UNABLE_TO_INFER_SCHEMA on the DSv2 path, " +
+ s"got ${e.getCondition}: ${e.getMessage}")
}
- assert(e.getCondition == "UNABLE_TO_INFER_SCHEMA",
- s"expected UNABLE_TO_INFER_SCHEMA, got ${e.getCondition}:
${e.getMessage}")
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
