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 26c11fdd3360 [SPARK-57195][SQL] Surface MALFORMED_CSV_RECORD instead
of ArrayIndexOutOfBoundsException in CSV schema inference
26c11fdd3360 is described below
commit 26c11fdd3360dbd253a72641b45db59d404d2c32
Author: Yash Botadra <[email protected]>
AuthorDate: Sun Jun 14 10:58:29 2026 -0700
[SPARK-57195][SQL] Surface MALFORMED_CSV_RECORD instead of
ArrayIndexOutOfBoundsException in CSV schema inference
### What changes were proposed in this pull request?
CSV schema inference can fail with an uncaught
`java.lang.ArrayIndexOutOfBoundsException` when a row has more columns than
`maxColumns`. [SPARK-49444](https://issues.apache.org/jira/browse/SPARK-49444)
added handling for the per-line `UnivocityParser.parseLine` path, but the
schema-inference paths that tokenize with a raw Univocity `CsvParser` were
never covered, so they still surface the internal exception.
This PR translates that exception into a `MALFORMED_CSV_RECORD` error
across the remaining paths:
- Adds a shared `UnivocityParser.parseLine(tokenizer, line)` helper that
converts Univocity's `ArrayIndexOutOfBoundsException` (raised bare or wrapped
in a `TextParsingException`) into `MALFORMED_CSV_RECORD`.
- The bad record in the message is bounded to
CSVOptions.MAX_ERROR_CONTENT_LENGTH
(the same bound Univocity uses).
- Guards the streaming tokenizer `UnivocityParser.convertStream`, used by
`multiLine` reads and `multiLine` schema inference.
- Routes the non-`multiLine` inference path
(`TextInputCSVDataSource.inferFromDataset`).
Note: Header rows are out of scope from this PR. A header over `maxColumns`
still surfaces the raw AIOOBE (`CSVHeaderChecker`), a pre-existing gap.
### Why are the changes needed?
Schema inference and `multiLine` reads crash with an internal
`ArrayIndexOutOfBoundsException` for input that should produce a clean
`MALFORMED_CSV_RECORD` error. SPARK-49444 only covered
`UnivocityParser.parseLine`; the inference paths construct a raw `CsvParser`
and call `parseLine` on it directly, bypassing that handling.
### Does this PR introduce _any_ user-facing change?
Yes.
- When a CSV row has more columns than `maxColumns` during schema inference
(or a `multiLine` read), the surfaced error changes from an internal
`java.lang.ArrayIndexOutOfBoundsException` to `MALFORMED_CSV_RECORD` (SQLSTATE
`KD000`), consistent with the non-`multiLine` per-row read path since
SPARK-49444. This is a change relative to all released versions.
- A maxCharsPerColumn overflow on the guarded paths also changes error
class (`TextParsingException` -> `MALFORMED_CSV_RECORD`), because the
direct-cause AIOOBE check can't distinguish it from `maxColumns`; the
bad-record content is bounded to 1000 chars.
### How was this patch tested?
Added unit tests in CSVSuite covering schema inference for the multiLine
and non-multiLine paths, plus the multiLine read path, asserting
MALFORMED_CSV_RECORD instead of a raw ArrayIndexOutOfBoundsException.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic), Claude Opus 4.8
Closes #56260 from yashtc/csv-infer-parsenext-malformed.
Lead-authored-by: Yash Botadra <[email protected]>
Co-authored-by: Yash Botadra <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit e1d39c6cb1bf851e2a2ac8c2ee28dbce1fde1822)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../apache/spark/sql/catalyst/csv/CSVOptions.scala | 6 +-
.../spark/sql/catalyst/csv/UnivocityParser.scala | 64 +++++++++++++----
.../execution/datasources/csv/CSVDataSource.scala | 5 +-
.../sql/execution/datasources/csv/CSVSuite.scala | 82 ++++++++++++++++++++++
4 files changed, 140 insertions(+), 17 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
index 4ae7aad19a9c..30b6877d3a82 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala
@@ -250,7 +250,7 @@ class CSVOptions(
/**
* The max error content length in CSV parser/writer exception message.
*/
- val maxErrorContentLength = 1000
+ val maxErrorContentLength = CSVOptions.MAX_ERROR_CONTENT_LENGTH
val isCommentSet = parameters.get(COMMENT) match {
case Some(value) if value.length == 1 => true
@@ -443,4 +443,8 @@ object CSVOptions extends DataSourceOptions {
newOption(SEP, DELIMITER)
val COLUMN_PRUNING = newOption("columnPruning")
val SINGLE_VARIANT_COLUMN =
newOption(DataSourceOptions.SINGLE_VARIANT_COLUMN)
+
+ // Max error content length in CSV parser/writer exception messages, and the
bound on the bad
+ // record embedded in MALFORMED_CSV_RECORD errors.
+ val MAX_ERROR_CONTENT_LENGTH = 1000
}
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 2073c5922fea..113f9b088738 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
@@ -314,19 +314,8 @@ class UnivocityParser(
}
}
- private def parseLine(line: String): Array[String] = {
- try {
- tokenizer.parseLine(line)
- }
- catch {
- case e: TextParsingException if
e.getCause.isInstanceOf[ArrayIndexOutOfBoundsException] =>
- throw new SparkRuntimeException(
- errorClass = "MALFORMED_CSV_RECORD",
- messageParameters = Map("badRecord" -> line),
- cause = e
- )
- }
- }
+ private def parseLine(line: String): Array[String] =
+ UnivocityParser.parseLine(tokenizer, line)
/**
* Parses a single CSV string and turns it into either one resulting row or
no row (if the
@@ -634,7 +623,19 @@ private[sql] object UnivocityParser {
// We can handle header here since here the stream is open.
handleHeader()
- private var nextRecord = tokenizer.parseNext()
+ // The bad-record text is not available here, so use the bounded parsed
content when present.
+ private def parseNextRecord(): Array[String] = {
+ try {
+ tokenizer.parseNext()
+ } catch {
+ case e: TextParsingException if
e.getCause.isInstanceOf[ArrayIndexOutOfBoundsException] =>
+ throw malformedCsvRecord(e, Option(e.getParsedContent).getOrElse(""))
+ case e: ArrayIndexOutOfBoundsException =>
+ throw malformedCsvRecord(e, "")
+ }
+ }
+
+ private var nextRecord = parseNextRecord()
override def hasNext: Boolean = nextRecord != null
@@ -643,11 +644,44 @@ private[sql] object UnivocityParser {
throw QueryExecutionErrors.endOfStreamError()
}
val curRecord = convert(nextRecord)
- nextRecord = tokenizer.parseNext()
+ nextRecord = parseNextRecord()
curRecord
}
}
+ /**
+ * Builds a MALFORMED_CSV_RECORD error from a Univocity malformed-record
failure. The bad record
+ * 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 = {
+ val boundedRecord = if (badRecord.length >
CSVOptions.MAX_ERROR_CONTENT_LENGTH) {
+ badRecord.take(CSVOptions.MAX_ERROR_CONTENT_LENGTH) + "..."
+ } else {
+ badRecord
+ }
+ new SparkRuntimeException(
+ errorClass = "MALFORMED_CSV_RECORD",
+ messageParameters = Map("badRecord" -> boundedRecord),
+ cause = cause)
+ }
+
+ /**
+ * Parses a single CSV line, translating a Univocity malformed-record
+ * ArrayIndexOutOfBoundsException into MALFORMED_CSV_RECORD so the per-line
and streaming paths
+ * fail consistently.
+ */
+ def parseLine(tokenizer: CsvParser, line: String): Array[String] = {
+ try {
+ tokenizer.parseLine(line)
+ } catch {
+ case e: TextParsingException if
e.getCause.isInstanceOf[ArrayIndexOutOfBoundsException] =>
+ throw malformedCsvRecord(e, line)
+ case e: ArrayIndexOutOfBoundsException =>
+ throw malformedCsvRecord(e, line)
+ }
+ }
+
/**
* Parses an iterator that contains CSV strings and turns it into an
iterator of rows.
*/
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 5170ce8a3da3..7424d43341b2 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
@@ -362,7 +362,10 @@ object TextInputCSVDataSource extends CSVDataSource {
val linesWithoutHeader =
CSVUtils.filterHeaderLine(filteredLines, maybeFirstLine.get,
parsedOptions)
val parser = new CsvParser(parsedOptions.asParserSettings)
- linesWithoutHeader.map(parser.parseLine)
+ // 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) {
new CSVInferSchema(parsedOptions).infer(tokenRDD, header)
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 22b291677cd8..e48b453309aa 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
@@ -3506,6 +3506,88 @@ abstract class CSVSuite
assert(textParsingException.getCause.isInstanceOf[ArrayIndexOutOfBoundsException])
}
+ test("SPARK-57195: multiLine CSV schema inference surfaces
MALFORMED_CSV_RECORD for a row " +
+ "exceeding maxColumns") {
+ // multiLine schema inference reads through
UnivocityParser.tokenizeStream, whose parseNext
+ // call was unguarded (SPARK-49444 only fixed the per-line parseLine
path). A row with more
+ // columns than maxColumns must surface as MALFORMED_CSV_RECORD, not a raw
+ // ArrayIndexOutOfBoundsException. The overflow is on a later row so it is
hit during inference.
+ withTempPath { path =>
+ Files.write(path.toPath,
"a,b\nc,d\n1,2,3\n".getBytes(StandardCharsets.UTF_8))
+ val e = intercept[SparkRuntimeException] {
+ spark.read
+ .option("header", "false")
+ .option("inferSchema", "true")
+ .option("multiLine", "true")
+ .option("maxColumns", "2")
+ .csv(path.getAbsolutePath)
+ }
+ // badRecord comes from TextParsingException.getParsedContent (bounded),
so its exact value
+ // is not pinned; ".*" keeps the error class, sqlState, and parameter
validation.
+ checkError(
+ exception = e,
+ condition = "MALFORMED_CSV_RECORD",
+ sqlState = Some("KD000"),
+ parameters = Map("badRecord" -> ".*"),
+ matchPVals = true)
+ }
+ }
+
+ test("SPARK-57195: non-multiLine CSV schema inference surfaces
MALFORMED_CSV_RECORD for a row " +
+ "exceeding maxColumns") {
+ // Without multiLine, inference reads through
TextInputCSVDataSource.inferFromDataset, which
+ // parsed each line with a raw Univocity CsvParser, bypassing the guarded
parseLine. A row with
+ // more columns than maxColumns must surface as MALFORMED_CSV_RECORD, not
a raw
+ // ArrayIndexOutOfBoundsException.
+ withTempPath { path =>
+ Files.write(path.toPath,
"a,b\nc,d\n1,2,3\n".getBytes(StandardCharsets.UTF_8))
+ val e = intercept[SparkRuntimeException] {
+ spark.read
+ .option("header", "false")
+ .option("inferSchema", "true")
+ .option("maxColumns", "2")
+ .csv(path.getAbsolutePath)
+ }
+ checkError(
+ exception = e,
+ condition = "MALFORMED_CSV_RECORD",
+ parameters = Map("badRecord" -> "1,2,3"),
+ sqlState = "KD000")
+ }
+ }
+
+ test("SPARK-57195: multiLine CSV read failure with more than max columns") {
+ // The multiLine read path (parseStream) uses the same guarded streaming
tokenizer as inference.
+ // With an explicit schema, an overflow row surfaces as
MALFORMED_CSV_RECORD wrapped in
+ // FAILED_READ_FILE, mirroring the non-multiLine SPARK-49444 test above.
+ val schema = new StructType()
+ .add("intColumn", IntegerType, nullable = true)
+ .add("decimalColumn", DecimalType(10, 2), nullable = true)
+
+ val fileReadException = intercept[SparkException] {
+ spark.read
+ .schema(schema)
+ .option("header", "false")
+ .option("multiLine", "true")
+ .option("maxColumns", "2")
+ .csv(testFile(moreColumnsFile))
+ .collect()
+ }
+
+ checkErrorMatchPVals(
+ exception = fileReadException,
+ condition = "FAILED_READ_FILE.NO_HINT",
+ parameters = Map("path" -> s".*$moreColumnsFile"))
+
+ val malformedCSVException =
fileReadException.getCause.asInstanceOf[SparkRuntimeException]
+ checkError(
+ exception = malformedCSVException,
+ condition = "MALFORMED_CSV_RECORD",
+ sqlState = Some("KD000"),
+ parameters = Map("badRecord" -> ".*"),
+ matchPVals = true)
+ }
+
test("csv with variant") {
withTempPath { path =>
val data =
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
