Skip to content

[SPARK-25387][SQL] Fix for NPE caused by bad CSV input #22374

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
MaxGekk wants to merge 7 commits into from
Closed

[SPARK-25387][SQL] Fix for NPE caused by bad CSV input #22374

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6f9aba5
Fix NPE in read with specified schema
MaxGekk Sep 9, 2018
9284527
Fix NPE in read on schema inferring
MaxGekk Sep 9, 2018
05fe5fa
Checking multiLine mode
MaxGekk Sep 9, 2018
b20c12d
Adding ticket number to test's title
MaxGekk Sep 9, 2018
c9ccbee
Fix imports
MaxGekk Sep 9, 2018
bd4ebe4
Test refactoring
MaxGekk Sep 10, 2018
2a0dac4
Refactoring of handling null from parseLine
MaxGekk Sep 11, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 19 additions & 17 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVDataSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,23 +240,25 @@ object TextInputCSVDataSource extends CSVDataSource {
sparkSession: SparkSession,
csv: Dataset[String],
maybeFirstLine: Option[String],
parsedOptions: CSVOptions): StructType = maybeFirstLine match {
case Some(firstLine) =>
val firstRow = new CsvParser(parsedOptions.asParserSettings).parseLine(firstLine)
val caseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
val header = makeSafeHeader(firstRow, caseSensitive, parsedOptions)
val sampled: Dataset[String] = CSVUtils.sample(csv, parsedOptions)
val tokenRDD = sampled.rdd.mapPartitions { iter =>
val filteredLines = CSVUtils.filterCommentAndEmpty(iter, parsedOptions)
val linesWithoutHeader =
CSVUtils.filterHeaderLine(filteredLines, firstLine, parsedOptions)
val parser = new CsvParser(parsedOptions.asParserSettings)
linesWithoutHeader.map(parser.parseLine)
}
CSVInferSchema.infer(tokenRDD, header, parsedOptions)
case None =>
// If the first line could not be read, just return the empty schema.
StructType(Nil)
parsedOptions: CSVOptions): StructType = {
val csvParser = new CsvParser(parsedOptions.asParserSettings)
maybeFirstLine.map(csvParser.parseLine(_)) match {
case Some(firstRow) if firstRow != null =>
val caseSensitive = sparkSession.sessionState.conf.caseSensitiveAnalysis
val header = makeSafeHeader(firstRow, caseSensitive, parsedOptions)
val sampled: Dataset[String] = CSVUtils.sample(csv, parsedOptions)
val tokenRDD = sampled.rdd.mapPartitions { iter =>
val filteredLines = CSVUtils.filterCommentAndEmpty(iter, parsedOptions)
val linesWithoutHeader =
CSVUtils.filterHeaderLine(filteredLines, maybeFirstLine.get, parsedOptions)
val parser = new CsvParser(parsedOptions.asParserSettings)
linesWithoutHeader.map(parser.parseLine)
}
CSVInferSchema.infer(tokenRDD, header, parsedOptions)
Copy link
Copy Markdown
Member

@HyukjinKwon HyukjinKwon Sep 13, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@MaxGekk, BTW what happen if the second line is the malfromed record and it returns null? From a cursory look, schema inference looks going to throw an NPE exception.

Copy link
Copy Markdown
Member Author

@MaxGekk MaxGekk Sep 13, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@HyukjinKwon I have checked this on (with header too):

    val input = spark.createDataset(Seq("1", "\u0000\u0000\u0001234"))

    val df = spark.read.option("inferSchema", true).csv(input)
    df.printSchema()
    df.show()
root
 |-- _c0: integer (nullable = true)

+----+
| _c0|
+----+
|   1|
|null|
+----+

In the debugger, I didn't observe null in

spark/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVInferSchema.scala

Lines 61 to 69 in 5264164

private def inferRowType(options: CSVOptions)
(rowSoFar: Array[DataType], next: Array[String]): Array[DataType] = {
var i = 0
while (i < math.min(rowSoFar.length, next.length)) { // May have columns on right missing.
rowSoFar(i) = inferField(rowSoFar(i), next(i), options)
i+=1
}
rowSoFar
}
.

case _ =>
// If the first line could not be read, just return the empty schema.
StructType(Nil)
}
}

private def createBaseDataset(
Expand Down
7 changes: 6 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/UnivocityParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,12 @@ class UnivocityParser(
}

private def convert(tokens: Array[String]): InternalRow = {
if (tokens.length != parsedSchema.length) {
if (tokens == null) {
Copy link
Copy Markdown
Contributor

@cloud-fan cloud-fan Sep 12, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when will we hit this?

Copy link
Copy Markdown
Member Author

@MaxGekk MaxGekk Sep 12, 2018 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I got it on a CSV file with some marks (a couple zero bytes) at the beginning but uniVocity parser returns null in many cases when it cannot read/parse input, for example: https://github.com/uniVocity/univocity-parsers/blob/f616d151b48150bc9cb98943f9b6f8353b704359/src/main/java/com/univocity/parsers/common/AbstractParser.java#L663

throw BadRecordException(
() => getCurrentInput,
() => None,
new RuntimeException("Malformed CSV record"))
} else if (tokens.length != parsedSchema.length) {
// If the number of tokens doesn't match the schema, we should treat it as a malformed record.
// However, we still have chance to parse some of the tokens, by adding extra null tokens in
// the tail if the number is smaller, or by dropping extra tokens if the number is larger.
Expand Down
11 changes: 10 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ import org.apache.log4j.{AppenderSkeleton, LogManager}
import org.apache.log4j.spi.LoggingEvent

import org.apache.spark.SparkException
import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row, UDT}
import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.{SharedSQLContext, SQLTestUtils}
Expand Down Expand Up @@ -1700,4 +1700,13 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils with Te
checkCount(2)
countForMalformedCSV(0, Seq(""))
}

test("SPARK-25387: bad input should not cause NPE") {
val schema = StructType(StructField("a", IntegerType) :: Nil)
val input = spark.createDataset(Seq("\u0000\u0000\u0001234"))
Copy link
Copy Markdown
Member

@maropu maropu Sep 10, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

btw, in this title, bad CSV means what (bad unicode?)? In this case, the CSV parser returns null and, in another case, it throws com.univocity.parsers.common.TextParsingException? I just want to know the behaivour in the parser.

Copy link
Copy Markdown
Member Author

@MaxGekk MaxGekk Sep 11, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The parseLine method can return null in many cases. See:
https://github.com/uniVocity/univocity-parsers/blob/f616d151b48150bc9cb98943f9b6f8353b704359/src/main/java/com/univocity/parsers/common/AbstractParser.java#L663
https://github.com/uniVocity/univocity-parsers/blob/f616d151b48150bc9cb98943f9b6f8353b704359/src/main/java/com/univocity/parsers/common/AbstractParser.java#L678

It is normal way for the method to indicate about an error.


checkAnswer(spark.read.schema(schema).csv(input), Row(null))
checkAnswer(spark.read.option("multiLine", true).schema(schema).csv(input), Row(null))
assert(spark.read.csv(input).collect().toSet == Set(Row()))
}
}