This is an automated email from the ASF dual-hosted git repository.
MaxGekk 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 5d9db4f3b8ba [SPARK-57515][SQL] Surface MALFORMED_CSV_RECORD instead
of ArrayIndexOutOfBoundsException when CSV header exceeds maxColumns
5d9db4f3b8ba is described below
commit 5d9db4f3b8bae4764b71657d3d8aff5d45906106
Author: Jubin Soni <[email protected]>
AuthorDate: Wed Jun 24 14:21:51 2026 +0200
[SPARK-57515][SQL] Surface MALFORMED_CSV_RECORD instead of
ArrayIndexOutOfBoundsException when CSV header exceeds maxColumns
### What is the purpose of the change?
Fixes [SPARK-57515](https://issues.apache.org/jira/browse/SPARK-57515).
When reading a CSV file with `header=true` and the header line has more columns
than `maxColumns`
(default 20480, user-configurable via `.option("maxColumns", N)`), Spark
crashes with an internal
`java.lang.ArrayIndexOutOfBoundsException` instead of a clean
`MALFORMED_CSV_RECORD` error.
[SPARK-57195](https://issues.apache.org/jira/browse/SPARK-57195) (merged
2026-06-14) fixed the same `ArrayIndexOutOfBoundsException` for data rows and
explicitly called out the remaining gap: _"Header rows are out of scope
from this PR. A header over
`maxColumns` still surfaces the raw AIOOBE (`CSVHeaderChecker`), a
pre-existing gap."_ This PR
closes that gap.
The bug affects all three CSV read paths that involve header parsing:
- **Non-multiLine file read** — `tokenizer.parseLine(header)` in
`CSVHeaderChecker.checkHeaderColumnNames(lines, tokenizer)` was called
directly, bypassing the AIOOBE guard.
- **MultiLine file read** — `tokenizer.parseNext()` in
`CSVHeaderChecker.checkHeaderColumnNames(tokenizer)` during header consumption
was unguarded.
- **`Dataset[String]` `csv()`** —
`CSVHeaderChecker.checkHeaderColumnNames(line: String)` created a fresh
`CsvParser` and called `parser.parseLine(line)` directly.
- **Schema inference (non-multiLine and `Dataset[String]`)** —
`CSVDataSource.inferFromDataset` parsed the first line (which is the header
when `header=true`) with a raw `CsvParser`, also bypassing the guard.
### Brief change log
- `CSVHeaderChecker.checkHeaderColumnNames(line: String)`: replaced
`parser.parseLine(line)` with `UnivocityParser.parseLine(parser, line)` to
reuse the existing safe wrapper from SPARK-57195.
- `CSVHeaderChecker.checkHeaderColumnNames(tokenizer)`: wrapped
`tokenizer.parseNext()` in a try/catch that translates
`ArrayIndexOutOfBoundsException` (bare or wrapped in `TextParsingException`)
into `MALFORMED_CSV_RECORD`.
- `CSVHeaderChecker.checkHeaderColumnNames(lines, tokenizer)`: wrapped
`tokenizer.parseLine(header)` in the same try/catch.
- `UnivocityParser.malformedCsvRecord` widened from `private` to
`private[csv]` so `CSVHeaderChecker` can reuse it directly, avoiding a
duplicate helper.
- `CSVDataSource.inferFromDataset`: replaced raw
`csvParser.parseLine(firstLine)` with `UnivocityParser.parseLine(csvParser,
firstLine)`, fixing the inference path for non-multiLine and `Dataset[String]`
reads. As a side effect, first-line `maxCharsPerColumn`/AIOOBE during inference
now also surfaces as `MALFORMED_CSV_RECORD` rather than a raw
`TextParsingException` (the SPARK-28431 test was updated to match).
### Verifying this change
Five new tests added to `CSVSuite`:
- **SPARK-57515: non-multiLine CSV read with header exceeding maxColumns
surfaces MALFORMED_CSV_RECORD** — no explicit schema; error surfaces from
`inferFromDataset` (the schema inference path).
- **SPARK-57515: non-multiLine CSV read with header exceeding maxColumns
and explicit schema surfaces MALFORMED_CSV_RECORD** — explicit schema skips
inference; error surfaces from `CSVHeaderChecker.checkHeaderColumnNames(lines,
tokenizer)`.
- **SPARK-57515: multiLine CSV read with header exceeding maxColumns
surfaces MALFORMED_CSV_RECORD** — same with `multiLine=true`; exercises
`CSVHeaderChecker.checkHeaderColumnNames(tokenizer)`.
- **SPARK-57515: Dataset[String] CSV read with header exceeding maxColumns
surfaces MALFORMED_CSV_RECORD** — no explicit schema; error surfaces from
`inferFromDataset` (the `Dataset[String]` inference path).
- **SPARK-57515: Dataset[String] CSV read with header exceeding maxColumns
and explicit schema surfaces MALFORMED_CSV_RECORD** — explicit schema skips
inference; error surfaces from `CSVHeaderChecker.checkHeaderColumnNames(line:
String)`.
### Does this PR potentially affect one of the following areas?
- Dependencies: no
- Public API: no — `CSVHeaderChecker` is internal
- Serializers: no
- Runtime per-record code paths (performance): no — only the header-parsing
path, which runs once per file
- Deployment or recovery: no
- S3 connector: no
### Documentation
This PR does not introduce a new feature. No documentation changes needed.
### Was generative AI tooling used to co-author this PR?
- [x] Yes — Claude Code was used as a pair-programming assistant. All code
was written, understood, and
verified by the author.
Generated-by: Claude Opus 4.8
Closes #56581 from jubins/j-SPARK-57515-csv-header-maxcolumn.
Authored-by: Jubin Soni <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../spark/sql/catalyst/csv/CSVHeaderChecker.scala | 30 ++++-
.../spark/sql/catalyst/csv/UnivocityParser.scala | 3 +-
.../execution/datasources/csv/CSVDataSource.scala | 5 +-
.../sql/execution/datasources/csv/CSVSuite.scala | 145 ++++++++++++++++++++-
4 files changed, 169 insertions(+), 14 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVHeaderChecker.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVHeaderChecker.scala
index bec52747dea7..ecd17077d557 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVHeaderChecker.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVHeaderChecker.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.catalyst.csv
-import com.univocity.parsers.common.AbstractParser
+import com.univocity.parsers.common.{AbstractParser, TextParsingException}
import com.univocity.parsers.csv.{CsvParser, CsvParserSettings}
import org.apache.spark.SparkIllegalArgumentException
@@ -122,7 +122,7 @@ class CSVHeaderChecker(
def checkHeaderColumnNames(line: String): Unit = {
if (options.headerFlag) {
val parser = new CsvParser(options.asParserSettings)
- checkHeaderColumnNames(parser.parseLine(line))
+ checkHeaderColumnNames(UnivocityParser.parseLine(parser, line))
}
}
@@ -130,7 +130,19 @@ class CSVHeaderChecker(
private[csv] def checkHeaderColumnNames(tokenizer:
AbstractParser[CsvParserSettings]): Unit = {
assert(options.multiLine, "This method should be executed with multiLine.")
if (options.headerFlag) {
- val firstRecord = tokenizer.parseNext()
+ val firstRecord = try {
+ tokenizer.parseNext()
+ } catch {
+ // scalastyle:off line.size.limit
+ case e: TextParsingException if
e.getCause.isInstanceOf[ArrayIndexOutOfBoundsException] =>
+ // scalastyle:on line.size.limit
+ // In the multiLine stream path the field appender is reset before
the AIOOBE propagates,
+ // so the record content is unavailable; use the bounded parsed
content when present,
+ // empty string as the fallback.
+ throw UnivocityParser.malformedCsvRecord(e,
Option(e.getParsedContent).getOrElse(""))
+ case e: ArrayIndexOutOfBoundsException =>
+ throw UnivocityParser.malformedCsvRecord(e, "")
+ }
checkHeaderColumnNames(firstRecord)
}
setHeaderForSingleVariantColumn.foreach(f => f(headerColumnNames))
@@ -146,7 +158,17 @@ class CSVHeaderChecker(
// be not extracted.
if (options.headerFlag && isStartOfFile) {
CSVExprUtils.extractHeader(lines, options).foreach { header =>
- checkHeaderColumnNames(tokenizer.parseLine(header))
+ val tokens = try {
+ tokenizer.parseLine(header)
+ } catch {
+ // scalastyle:off line.size.limit
+ case e: TextParsingException if
e.getCause.isInstanceOf[ArrayIndexOutOfBoundsException] =>
+ // scalastyle:on line.size.limit
+ throw UnivocityParser.malformedCsvRecord(e, header)
+ case e: ArrayIndexOutOfBoundsException =>
+ throw UnivocityParser.malformedCsvRecord(e, header)
+ }
+ checkHeaderColumnNames(tokens)
}
}
setHeaderForSingleVariantColumn.foreach(f => f(headerColumnNames))
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
index 113f9b088738..a028f77495a4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala
@@ -654,7 +654,8 @@ private[sql] object UnivocityParser {
* is bounded to CSVOptions.MAX_ERROR_CONTENT_LENGTH so an oversized value
cannot produce a huge
* error message (SPARK-28431).
*/
- private def malformedCsvRecord(cause: Throwable, badRecord: String):
SparkRuntimeException = {
+ private[csv] def malformedCsvRecord(
+ cause: Throwable, badRecord: String): SparkRuntimeException = {
val boundedRecord = if (badRecord.length >
CSVOptions.MAX_ERROR_CONTENT_LENGTH) {
badRecord.take(CSVOptions.MAX_ERROR_CONTENT_LENGTH) + "..."
} else {
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 10771114bc70..2bdc47e8e2da 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
@@ -359,7 +359,7 @@ object TextInputCSVDataSource extends CSVDataSource {
maybeFirstLine: Option[String],
parsedOptions: CSVOptions): StructType = {
val csvParser = new CsvParser(parsedOptions.asParserSettings)
- maybeFirstLine.map(csvParser.parseLine(_)) match {
+ maybeFirstLine.map(UnivocityParser.parseLine(csvParser, _)) match {
case Some(firstRow) if firstRow != null =>
val caseSensitive =
sparkSession.sessionState.conf.caseSensitiveAnalysis
val header = CSVUtils.makeSafeHeader(firstRow, caseSensitive,
parsedOptions)
@@ -369,9 +369,6 @@ object TextInputCSVDataSource extends CSVDataSource {
val linesWithoutHeader =
CSVUtils.filterHeaderLine(filteredLines, maybeFirstLine.get,
parsedOptions)
val parser = new CsvParser(parsedOptions.asParserSettings)
- // Route data rows through UnivocityParser.parseLine so a
too-many-columns row surfaces as
- // MALFORMED_CSV_RECORD, not a raw ArrayIndexOutOfBoundsException
(SPARK-57195). The
- // first-line parse above stays raw to keep SPARK-28431's bounded
TextParsingException.
linesWithoutHeader.map(UnivocityParser.parseLine(parser, _))
}
SQLExecution.withSQLConfPropagated(csv.sparkSession) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
index 0144c652b688..71eb34134920 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
@@ -2610,15 +2610,20 @@ abstract class CSVSuite
StandardOpenOption.CREATE, StandardOpenOption.WRITE
)
- val errMsg = intercept[TextParsingException] {
+ // Univocity wraps maxCharsPerColumn violations as
TextParsingException(cause=AIOOBE),
+ // which UnivocityParser.parseLine now converts to MALFORMED_CSV_RECORD.
The badRecord
+ // value is still bounded to MAX_ERROR_CONTENT_LENGTH (1000 chars) with
a "..." suffix,
+ // preserving the original intent of SPARK-28431.
+ val e = intercept[SparkRuntimeException] {
spark.read
.option("maxCharsPerColumn", maxCharsPerCol)
.csv(path.getAbsolutePath)
.count()
- }.getMessage
-
- assert(errMsg.contains("..."),
- "expect the TextParsingException truncate the error content to be 1000
length.")
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "MALFORMED_CSV_RECORD",
+ parameters = Map("badRecord" -> ".*\\.\\.\\."))
}
}
@@ -3588,6 +3593,136 @@ abstract class CSVSuite
matchPVals = true)
}
+ test("SPARK-57515: non-multiLine CSV read with header exceeding maxColumns
surfaces " +
+ "MALFORMED_CSV_RECORD") {
+ // Without an explicit schema, inference runs eagerly via inferFromDataset
and parses the first
+ // line there, so the error surfaces from CSVDataSource rather than from
CSVHeaderChecker.
+ // This test validates the inferFromDataset path.
+ withTempPath { path =>
+ Files.write(path.toPath,
"a,b,c\n1,2,3\n".getBytes(StandardCharsets.UTF_8))
+ val e = intercept[SparkRuntimeException] {
+ spark.read
+ .option("header", "true")
+ .option("maxColumns", "2")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkError(
+ exception = e,
+ condition = "MALFORMED_CSV_RECORD",
+ sqlState = Some("KD000"),
+ parameters = Map("badRecord" -> "a,b,c"),
+ matchPVals = false)
+ }
+ }
+
+ test("SPARK-57515: non-multiLine CSV read with header exceeding maxColumns
and explicit schema " +
+ "surfaces MALFORMED_CSV_RECORD") {
+ // With an explicit schema, inference is skipped and the read-time header
check in
+ // CSVHeaderChecker.checkHeaderColumnNames(lines, tokenizer) runs inside a
Spark task, so the
+ // SparkRuntimeException(MALFORMED_CSV_RECORD) is wrapped in
SparkException(FAILED_READ_FILE).
+ // Verify the cause chain surfaces the MALFORMED_CSV_RECORD condition.
+ withTempPath { path =>
+ Files.write(path.toPath,
"a,b,c\n1,2,3\n".getBytes(StandardCharsets.UTF_8))
+ val schema = StructType(Seq(
+ StructField("a", StringType), StructField("b", StringType)))
+ val e = intercept[SparkException] {
+ spark.read
+ .schema(schema)
+ .option("header", "true")
+ .option("maxColumns", "2")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> ".*"))
+ val cause = e.getCause
+ assert(cause.isInstanceOf[SparkRuntimeException])
+ checkError(
+ exception = cause.asInstanceOf[SparkRuntimeException],
+ condition = "MALFORMED_CSV_RECORD",
+ sqlState = Some("KD000"),
+ parameters = Map("badRecord" -> "a,b,c"),
+ matchPVals = false)
+ }
+ }
+
+ test("SPARK-57515: multiLine CSV read with header exceeding maxColumns
surfaces " +
+ "MALFORMED_CSV_RECORD") {
+ // For multiLine reads, schema inference runs inside an RDD task, so the
+ // SparkRuntimeException(MALFORMED_CSV_RECORD) is wrapped in
SparkException(FAILED_READ_FILE).
+ // Verify the cause chain surfaces the MALFORMED_CSV_RECORD condition.
+ withTempPath { path =>
+ Files.write(path.toPath,
"a,b,c\n1,2,3\n".getBytes(StandardCharsets.UTF_8))
+ val e = intercept[SparkException] {
+ spark.read
+ .option("header", "true")
+ .option("multiLine", "true")
+ .option("maxColumns", "2")
+ .csv(path.getAbsolutePath)
+ .collect()
+ }
+ checkErrorMatchPVals(
+ exception = e,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> ".*"))
+ val cause = e.getCause
+ assert(cause.isInstanceOf[SparkRuntimeException])
+ // In the multiLine path the header is parsed from a live stream via
parseNext(); by the time
+ // the AIOOBE is caught the field appender has already been reset, so
badRecord is empty.
+ checkErrorMatchPVals(
+ exception = cause.asInstanceOf[SparkRuntimeException],
+ condition = "MALFORMED_CSV_RECORD",
+ parameters = Map("badRecord" -> ".*"))
+ }
+ }
+
+ test("SPARK-57515: Dataset[String] CSV read with header exceeding maxColumns
surfaces " +
+ "MALFORMED_CSV_RECORD") {
+ // Without an explicit schema, inference runs eagerly via inferFromDataset
and parses the
+ // first line there. This validates the inferFromDataset path for
Dataset[String].
+ val lines = spark.createDataset(Seq("a,b,c", "1,2,3"))
+ val e = intercept[SparkRuntimeException] {
+ spark.read
+ .option("header", "true")
+ .option("maxColumns", "2")
+ .csv(lines)
+ .collect()
+ }
+ checkError(
+ exception = e,
+ condition = "MALFORMED_CSV_RECORD",
+ sqlState = Some("KD000"),
+ parameters = Map("badRecord" -> "a,b,c"),
+ matchPVals = false)
+ }
+
+ test("SPARK-57515: Dataset[String] CSV read with header exceeding maxColumns
and explicit " +
+ "schema surfaces MALFORMED_CSV_RECORD") {
+ // With an explicit schema, inference is skipped and the read-time header
check in
+ // CSVHeaderChecker.checkHeaderColumnNames(line: String) runs. That guard
must surface
+ // MALFORMED_CSV_RECORD rather than a raw TextParsingException.
+ val schema = StructType(Seq(
+ StructField("a", StringType), StructField("b", StringType)))
+ val lines = spark.createDataset(Seq("a,b,c", "1,2,3"))
+ val e = intercept[SparkRuntimeException] {
+ spark.read
+ .schema(schema)
+ .option("header", "true")
+ .option("maxColumns", "2")
+ .csv(lines)
+ .collect()
+ }
+ checkError(
+ exception = e,
+ condition = "MALFORMED_CSV_RECORD",
+ sqlState = Some("KD000"),
+ parameters = Map("badRecord" -> "a,b,c"),
+ matchPVals = false)
+ }
+
test("csv with variant") {
withTempPath { path =>
val data =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
