[SPARK-30530][SQL] Fix filter pushdown for bad CSV records

[MaxGekk]

What changes were proposed in this pull request?

In the PR, I propose to fix the bug reported in SPARK-30530. CSV datasource returns invalid records in the case when parsedSchema is shorter than number of tokens returned by UniVocity parser. In the case UnivocityParser.convert() always throws BadRecordException independently from the result of applying filters.

For the described case, I propose to save the exception in badRecordException and continue value conversion according to parsedSchema. If a bad record doesn't pass filters, convert() returns empty Seq otherwise throws badRecordException.

Why are the changes needed?

It fixes the bug reported in the JIRA ticket.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added new test from the JIRA ticket.

[SparkQA]

Test build #116867 has finished for PR 27239 at commit fd48094.

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

[cloud-fan]

if the first column is corrupted, and the predicate is first_col is null, what will happen?

[MaxGekk]

There are 3 cases:

1. Univocity parser is not able to parse its input. For example, it faced to wrong Unicode symbol. In that case, it return null in tokens, and BadRecordException will be raised here

   spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

   Lines 229 to 232 in 4e50f02

   	throw BadRecordException(
   	() => getCurrentInput,
   	() => None,
   	new RuntimeException("Malformed CSV record"))

2. Univocity parser returns null in the first token. In this case, we will try to convert null to desired type according to requiredSchema. Most likely, the conversion raises an exception which is will be converted BadRecordException here

   spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

   Line 277 in 4e50f02

   badRecordException = badRecordException.orElse(Some(e))

   and here

   spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

   Line 286 in 4e50f02

   throw BadRecordException(

   
   2.1 If conversion doesn't fail, the is null filter will be applied to the value and row could be passed to upper layer.
3. Univocity parser returns a valid string at index 0 in tokens but conversion fails at

   spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

   Line 267 in 4e50f02

   row(i) = valueConverters(i).apply(getToken(tokens, i))

   with some exception. Similar situation to 2. The exception will be handled, and transformed to BadRecordException.

New implementation with filters pushdown does not change the behavior in those cases.

[cloud-fan]

thanks, merging to master!

[HyukjinKwon]

Do we need this checkedTokens now?

[MaxGekk]

It seems not. The if can be replaced by:

    var badRecordException: Option[Throwable] = if (tokens.length != parsedSchema.length) {
      // If the number of tokens doesn't match the schema, we should treat it as a malformed record.
      Some(new RuntimeException("Malformed CSV record"))
    } else None

Let me do that in a follow up PR.

[HyukjinKwon]

Oops, I already did at #27287. Let me address this comment there.

[HyukjinKwon]

nit:

if (skipRow) {
  row.setNullAt(i)
} else {
  row(i) = valueConverters(i).apply(getToken(tokens, i))
  if (csvFilters.skipRow(row, i)) {
    skipRow = true
  }
}

[HyukjinKwon]

Previously we rely on nulls already exiting in the array. Now we rely on java.lang.ArrayIndexOutOfBoundsException. I don't particularly like this approach .. but I'm good as it does simplify the codes.