[SPARK-45035][SQL] Fix ignoreCorruptFiles/ignoreMissingFiles with multiline CSV/JSON will report error

[Hisoka-X]

What changes were proposed in this pull request?

Fix ignoreCorruptFiles/ignoreMissingFiles with multiline CSV/JSON will report error, it would be like:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4940.0 failed 4 times, most recent failure: Lost task 0.3 in stage 4940.0 (TID 4031) (10.68.177.106 executor 0): com.univocity.parsers.common.TextParsingException: java.lang.IllegalStateException - Error reading from input
Parser Configuration: CsvParserSettings:
	Auto configuration enabled=true
	Auto-closing enabled=true
	Autodetect column delimiter=false
	Autodetect quotes=false
	Column reordering enabled=true
	Delimiters for detection=null
	Empty value=
	Escape unquoted values=false
	Header extraction enabled=null
	Headers=null
	Ignore leading whitespaces=false
	Ignore leading whitespaces in quotes=false
	Ignore trailing whitespaces=false
	Ignore trailing whitespaces in quotes=false
	Input buffer size=1048576
	Input reading on separate thread=false
	Keep escape sequences=false
	Keep quotes=false
	Length of content displayed on error=1000
	Line separator detection enabled=true
	Maximum number of characters per column=-1
	Maximum number of columns=20480
	Normalize escaped line separators=true
	Null value=
	Number of records to read=all
	Processor=none
	Restricting data in exceptions=false
	RowProcessor error handler=null
	Selected fields=none
	Skip bits as whitespace=true
	Skip empty lines=true
	Unescaped quote handling=STOP_AT_DELIMITERFormat configuration:
	CsvFormat:
		Comment character=#
		Field delimiter=,
		Line separator (normalized)=\n
		Line separator sequence=\n
		Quote character="
		Quote escape character=\
		Quote escape escape character=null
Internal state when error was thrown: line=0, column=0, record=0
	at com.univocity.parsers.common.AbstractParser.handleException(AbstractParser.java:402)
	at com.univocity.parsers.common.AbstractParser.beginParsing(AbstractParser.java:277)
	at com.univocity.parsers.common.AbstractParser.beginParsing(AbstractParser.java:843)
	at org.apache.spark.sql.catalyst.csv.UnivocityParser$$anon$1.<init>(UnivocityParser.scala:463)
	at org.apache.spark.sql.catalyst.csv.UnivocityParser$.convertStream(UnivocityParser.scala:46... 

Because multiline CSV/JSON use BinaryFileRDD not FileScanRDD. Unlike FileScanRDD, when met corrupt files will check ignoreCorruptFiles config to avoid report IOException, BinaryFileRDD will not report error because it return normal PortableDataStream. So we should catch it when infer schema in lambda function. Also do same thing for ignoreMissingFiles.

Why are the changes needed?

Fix the bug when use mulitline mode with ignoreCorruptFiles/ignoreMissingFiles config.

Does this PR introduce any user-facing change?

No

How was this patch tested?

add new test.

Was this patch authored or co-authored using generative AI tooling?

No

[Hisoka-X]

cc @HyukjinKwon

[MaxGekk]

@Hisoka-X Could you rebase this on the recent master Scala 2.13 + Java 17.

[Hisoka-X]

@Hisoka-X Could you rebase this on the recent master Scala 2.13 + Java 17.

Done

[MaxGekk]

How about RuntimeException like at

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

Line 262 in 7796d8a

case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>

?

[Hisoka-X]

RuntimeException already be catched in https://github.com/apache/spark/pull/42979/files/942b7c38a277a1b38b85a64ad940b56528ae8a03#diff-774d08eb04cd18039c576c7e23609430476d3dd2668535f0432f04b65b8ab234R92.

[MaxGekk]

I feel like JSON/CSV infer code should be similar to FileScanRDD and ignoreCorruptFiles should impact on errors caused by RuntimeException. What happens if put RuntimeException in the case? :

                  case e @ (_: RuntimeException | _: IOException) if ignoreCorruptFiles =>

[Hisoka-X]

It never reach here if we catch one RuntimeException. Because it will be catched by https://github.com/apache/spark/pull/42979/files/942b7c38a277a1b38b85a64ad940b56528ae8a03#diff-774d08eb04cd18039c576c7e23609430476d3dd2668535f0432f04b65b8ab234R92. So it would be useless code.

[MaxGekk]

I mean remove it from the case:

          case e @ (_: JsonProcessingException | _: MalformedInputException) =>
            handleJsonErrorsByParseMode(parseMode, columnNameOfCorruptRecord, e)
...

and put it there:

          case e @ (_: IOException | _: RuntimeException) if options.ignoreCorruptFiles =>
            logWarning("Skipped the rest of the content in the corrupted file", e)
            Some(StructType(Nil))

[Hisoka-X]

oh, I got it. But I'm not sure it would change behavior or not. Let me change it and see CI would report error or not.

[MaxGekk]

From my understanding, we consider a JSON record as malformed when the JSON parser cannot parse its already retrieved input. So, it was able to read some text from a file but cannot parse. For instance, if we look at JSON parser exceptions:

    public JsonParser createParser(InputStream in) throws IOException, JsonParseException {
        IOContext ctxt = _createContext(_createContentReference(in), false);
        return _createParser(_decorate(in, ctxt), ctxt);
    }
...
    public abstract JsonToken nextToken() throws IOException;

JsonParseException (JsonProcessingException) + IOException

So, the RuntimeException can come only for some corrupted files. cc @HyukjinKwon @cloud-fan

[MaxGekk]

Please, adjust PR's title and its description regarding to ignoreMissingFiles .

[cloud-fan]

It's a bit annoying that we need to handle corrupted/missing files in multiple places, but I don't have a better idea.

[MaxGekk]

+1, LGTM. Merging to master.
Thank you, @Hisoka-X and @cloud-fan @HyukjinKwon @beliefer for review.