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

[MaxGekk]

What changes were proposed in this pull request?

The PR fixes NPE in UnivocityParser caused by malformed CSV input. In some cases, uniVocity parser can return null for bad input. In the PR, I propose to check result of parsing and not propagate NPE to upper layers.

How was this patch tested?

I added a test which reproduce the issue and tested by CSVSuite.

[SparkQA]

Test build #95848 has finished for PR 22374 at commit c9ccbee.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

This line below possibly returns null?

spark/sql/core/src/main/scala/org/apache/spark/sql/DataFrameReader.scala

Line 510 in 77c9964

val columnNames = parser.parseLine(firstLine)

[MaxGekk]

This line below possibly returns null?

@maropu It can return null but inside of CSVDataSource.checkHeaderColumnNames there is a null checking.

[maropu]

ok, thanks for the check!

[maropu]

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.

[MaxGekk]

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.

[SparkQA]

Test build #95904 has finished for PR 22374 at commit bd4ebe4.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[gengliangwang]

Can we simplify the code as

    maybeFirstLine.map(new CsvParser(parsedOptions.asParserSettings).parseLine(_)) match {
      case Some(firstRow) if firstRow != null =>
      case _ =>

[SparkQA]

Test build #95939 has finished for PR 22374 at commit 2a0dac4.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[cloud-fan]

when will we hit this?

[MaxGekk]

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

[cloud-fan]

thanks, merging to master/2.4!

[HyukjinKwon]

LGTM too

[HyukjinKwon]

@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.

[MaxGekk]

@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
	}

.