[COLDMANYU]

Purpose

Closes #10773.

Spark's CSV source drops lines that are empty or contain only whitespace before parsing them. Paimon's CSV reader reads such a line as a row where every column is NULL. When an existing Spark CSV table is moved to a Paimon CSV Format Table, the same files return extra all-NULL rows. In the case described in the issue, one hour of tab-separated gzip logs returned 6 extra rows.

Blank lines cannot be skipped unconditionally. Paimon's own CsvFormatWriter writes a row whose values are all NULL as an empty line in a single-column table, and today such a line is read back correctly (see the table in #10773). This PR therefore adds an opt-in option:

csv.skip-empty-lines = false   (default: current behavior)

When it is enabled, CsvFileReader skips a line if nothing is left after String#trim. Spark uses the same test (CSVExprUtils.filterCommentAndEmpty: iter.filter(_.trim.nonEmpty)), so tables moved from Spark return the same rows.

Notes:

• The issue suggested StringUtils#isNullOrWhitespaceOnly. I used String#trim instead so that the result matches Spark exactly. The two only differ for a few characters. String#trim also removes control characters up to U+0020, so a line holding only Hive's default delimiter \u0001 (an all-NULL row) is skipped, as in Spark. Unicode whitespace above U+0020, such as U+3000, is kept. I'm happy to switch if you prefer the other one.
• The default behavior does not change. When the option is disabled, nothing is skipped.
• The option is only supported for format tables. SchemaValidation rejects it for other table types, including when it is set with ALTER TABLE or passed as a dynamic option. A file-store table records the number of written rows in its file metadata, and COUNT(*) push-down reads that number, so skipping rows while scanning would make the two disagree. Format table splits report no row count, so they always scan the rows.
• The header line (csv.include-header) is still handled before any line is skipped. Splitting by offset and length is not affected, because skipping happens per line after the line reader.
• The option is documented in the CSV format options table.

Tests

New CsvSkipEmptyLinesTest:

• testDefaultReadsBlankLinesAsNullRows: without the option, empty, space-only and tab-only lines are still read as all-NULL rows.
• testSkipEmptyLines: with the option, these lines are skipped and all other rows are returned in order.
• testSkipEmptyLinesInGzipFile: the same for a gzip file.
• testSkipEmptyLinesWithHeader: with csv.include-header, the header is still skipped and blank lines after it are skipped too.
• testSkipEmptyLinesWithHiveDefaultDelimiter: a line holding only \u0001 is skipped, as in Spark.

New CsvSkipEmptyLinesTableTest:

• Creating a file-store table with the option, setting it with ALTER TABLE, and passing it as a dynamic option all fail.
• A single-column CSV file-store table with rows (NULL), (1) returns 2 rows from a scan, and the metadata row count is also 2.
• A CSV format table with the option skips empty lines, and its splits report no row count.

The four CsvSkipEmptyLinesTest tests that enable the option fail without the change in CsvFileReader, and the three rejection tests fail without the check in SchemaValidation.

[JingsongLi]

The external CSV Format Table use case is supported, and the CSV suite passes locally (61 passed, 2 skipped; all 5 new tests pass). The finding below is backed by real table writes and scans. The Spark metadata-aggregation consumer was traced in code; a Spark SQL query was not run locally.

[JingsongLi]

[P2] Keep persisted-table metadata counts consistent when skipping rows

This option is also accepted by a normal append-only FileStoreTable. I reproduced this with a nullable single INT column, file.format=csv, file.compression=gzip, bucket=-1, and csv.skip-empty-lines=true: write (NULL), (1), commit, and scan. The reader returns 1 row, but the planned DataSplit still reports mergedRowCount() = 2. With the option disabled, both counts are 2. Spark's AggregatePushDownUtils.tryPushdownAggregation accepts this split and CountStarEvaluator sums that metadata, so a pushed COUNT(*) reports 2 although a normal scan returns only 1. Please restrict this behavior to external FormatTable reads, or reject/handle it for persisted tables and disable metadata aggregation wherever rows can be skipped. Add a table-level regression covering the count versus scanned rows.

[COLDMANYU]

Thanks for the careful review and the reproduction! You're right, I only had external format tables in mind and missed that a normal file-store table accepts the option too.

I chose to reject it for persisted tables. Our use case only needs it for external format tables that read CSV files written by Spark, so restricting the option keeps the change small and avoids touching metadata aggregation. If there is a use case for file-store tables later, it could be supported then by disabling metadata aggregation wherever rows can be skipped.

• SchemaValidation.validateTableSchema now rejects csv.skip-empty-lines=true unless the table type is format-table. This check also runs when an ALTER TABLE sets the option and when the option is passed as a dynamic option through Table#copy, so none of these paths can turn it on for a file-store table.
• Format tables are not affected by the count issue: FormatDataSplit#mergedRowCount() is always empty, so AggregatePushDownUtils.tryPushdownAggregation never pushes COUNT(*) down for them and the rows are always scanned.
• The option description and the docs now say that it is only supported for format tables.

New table-level tests in CsvSkipEmptyLinesTableTest:

• Creating a file-store table with the option, setting it with ALTER TABLE, and passing it as a dynamic option all fail.
• With your setup (single nullable INT column, csv, gzip, bucket=-1, rows (NULL), (1)), a scan without the option returns 2 rows and the metadata row count is also 2.
• A CSV format table with the option skips the empty lines, and its splits report no row count.

Without the new check, the three rejection tests fail. A scan with the option then returned 1 row while the metadata still reported 2, the same mismatch you found.