[COLDMANYU]

Purpose

scan.ignore-corrupt-files misses two ways a Format Table file can be corrupt, so one bad file still fails the whole query:

1. The file fails while rows are being read. Row-based formats (CSV, JSON, TEXT, Avro) return a lazy iterator from readBatch() and only decode rows in next(). When a file is corrupt in the middle, the error is thrown after the point where the option catches it. A typical case is a truncated gzip file: it opens normally, and java.io.EOFException: Unexpected end of input stream is thrown only when decompression reaches the cut.
2. The file fails while it is being opened. FormatReadBuilder creates its readers itself, so it never goes through the open-time handling in DataFileRecordReader. Text readers fill their first buffer in the constructor, so an empty or very small truncated gzip file already fails at this step.

Spark's spark.sql.files.ignoreCorruptFiles does not help in either case. A Paimon Format Table is read by Paimon's own reader, not by Spark's file scan.

Changes:

• DataFileRecordReader: when scan.ignore-corrupt-files is enabled, the iterator it returns is also guarded. If next() or skip() throws an exception that the option already treats as corruption, the rows returned so far are kept, a warning is logged, and the rest of the file is skipped. This matches the semantics of Spark's ignoreCorruptFiles.
• FormatReadBuilder: when the option is enabled and the reader cannot be created for a file that exists, a warning is logged and the file is skipped. A missing file is still reported as FileNotFoundException; scan.ignore-lost-files still handles that case.

Notes:

• When the option is disabled (the default), behavior does not change and the iterator is not wrapped.
• Columnar iterators (ColumnarRowIterator, used by Parquet, ORC and similar formats) are left unwrapped. They read the whole batch in readBatch(), which the option already covers, and callers check their type to detect vectorized batches.
• The set of exceptions treated as corruption does not change.

I also tested this on a Spark 3.5 cluster with one real hour of gzip text logs: 20 files, 7 of them truncated mid-stream and 1 empty. Without the change, the query fails with EOFException. With the change, it completes and logs one warning per corrupt file.

Tests

New FormatTableIgnoreCorruptFilesTest, parameterized over CSV and JSON:

• testTruncatedCompressedFile: one gzip file truncated in the middle plus one intact file. Without the option, the read fails with EOFException. With the option, it returns the rows before the cut and all rows of the intact file.
• testCompressedFileCorruptAtOpen: one empty gzip file, one gzip file cut inside its header, and one intact file. Without the option, the read fails. With the option, it returns all rows of the intact file.
• testLostFileIsNotTreatedAsCorrupt: a file deleted after planning still fails with FileNotFoundException when only scan.ignore-corrupt-files is enabled.

The first two tests fail on master and pass with this change.

[Akash3121]

ignoreCorruptException includes InternalError (notably ZipError), but this catch only intercepts Exception, so that supported corrupt-file failure still escapes and aborts the scan. Could we catch the same throwable set here (and in skip()), with a regression test using ZipError?

[COLDMANYU]

Good catch, thanks! InternalError was never reachable there: every catch site only caught Exception, so the InternalError branch of ignoreCorruptException was dead code. This was already true of the two existing sites in createReader() and readBatch(), and I copied the same pattern into the new ones.

I've fixed it in all of them:

• DataFileRecordReader: createReader(), readBatch(), and the iterator's next() and skip() now catch Exception | InternalError.
• FormatReadBuilder: the reader creation catch does the same.

InternalError is only caught so that it can be ignored. When scan.ignore-corrupt-files is disabled, it is rethrown unchanged, so the default behavior stays the same. Other Errors, such as OutOfMemoryError, are still not caught.

This also matches Spark. Since SPARK-52482 (4.1.0), FileScanRDD uses DataSourceUtils.shouldIgnoreCorruptFileException, which treats RuntimeException | IOException | InternalError as corruption. That change was made for the same reason: a corrupt zstd header throws java.lang.InternalError: Unknown frame descriptor.

I added testZipErrorWhileReading and testZipErrorAtOpen (CSV and JSON). They use a FileIO that throws ZipError either while the file is being read or when it is opened. Without the option, the ZipError propagates. With the option, the corrupt file is skipped and the intact file is still read completely. Both tests fail without this change.

[JingsongLi]

+1