[format] Add csv.skip-empty-lines to skip empty lines when reading CSV

docs/docs/concepts/spec/fileformat.md

@@ -155,6 +155,7 @@ Format Options:
 | `csv.escape-character` | `\` | String | The escape character for CSV format, must be single character. |
 | `csv.include-header` | false | Boolean | Whether to include header in CSV files. |
 | `csv.null-literal` | `""` | String | Null literal string that is interpreted as a null value (disabled by default). |
+| `csv.skip-empty-lines` | false | Boolean | Whether to skip empty lines when reading, the same way as Spark's CSV source. Only supported for format tables. A line is empty if nothing is left after `String#trim`, which removes spaces, tabs and other characters up to U+0020. By default such a line is read as a row of NULLs. Paimon itself writes a row whose values are all NULL as an empty line in a single-column table, or as a line of delimiters only (e.g. a tab), so such rows are also skipped when this option is enabled. |
 | `csv.mode` | `PERMISSIVE` | String | Allows a mode for dealing with corrupt records during reading. Currently supported values are `'PERMISSIVE'`, `'DROPMALFORMED'` and `'FAILFAST'`: <ul> <li>Option `'PERMISSIVE'` sets malformed fields to null.</li> <li>Option `'DROPMALFORMED'` ignores the whole corrupted records.</li> <li>Option `'FAILFAST'` throws an exception when it meets corrupted records.</li> </ul> |
 
 Paimon CSV format uses [jackson databind API](https://github.com/FasterXML/jackson-databind) to parse and generate CSV string.

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -29,6 +29,7 @@
 import org.apache.paimon.fileindex.FileIndexerFactory;
 import org.apache.paimon.fileindex.FileIndexerFactoryUtils;
 import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.csv.CsvOptions;
 import org.apache.paimon.globalindex.GlobalIndexer;
 import org.apache.paimon.globalindex.bitmap.BitmapGlobalIndexerFactory;
 import org.apache.paimon.globalindex.bitmap.MultiValueGlobalIndexerFactory;
@@ -176,6 +177,7 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
         // reject here rather than only at create time, so ALTER cannot turn the option on for a
         // table type that ignores it
         validateQueryAuthTableType(options.type(), options.queryAuthEnabled());
+        validateCsvSkipEmptyLines(options);
 
         if (options.primaryKeyNullable() && schema.primaryKeys().isEmpty()) {
             throw new IllegalArgumentException(
@@ -445,6 +447,21 @@ public static void validateTableSchema(TableSchema schema, Set<String> dynamicOp
         validateManifestSort(schema, options);
     }
 
+    /**
+     * Skipping empty lines drops rows at read time, while the data files of a table still record
+     * the number of rows that were written. Readers that answer COUNT(*) from that metadata would
+     * then disagree with a normal scan, so the option is only allowed for format tables, whose
+     * splits do not report row counts.
+     */
+    private static void validateCsvSkipEmptyLines(CoreOptions options) {
+        checkArgument(
+                options.type() == TableType.FORMAT_TABLE
+                        || !options.toConfiguration().get(CsvOptions.SKIP_EMPTY_LINES),
+                "%s is only supported for format tables, but the table type is %s.",
+                CsvOptions.SKIP_EMPTY_LINES.key(),
+                options.type());
+    }
+
     /**
      * Only a file-store table reads through the auth reader; anywhere else the rules would be
      * accepted and then silently not applied.

paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java

@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.csv.CsvOptions;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.testutils.assertj.PaimonAssertions.anyCauseMatches;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests that {@link CsvOptions#SKIP_EMPTY_LINES} is only accepted for format tables. A file-store
+ * table records the number of written rows in its file metadata, which COUNT(*) push-down reads, so
+ * skipping rows while scanning would make the two disagree.
+ */
+class CsvSkipEmptyLinesTableTest extends TableTestBase {
+
+    private static final String ERROR = "csv.skip-empty-lines is only supported for format tables";
+
+    @Test
+    void testRejectedWhenCreatingFileStoreTable() {
+        assertThatThrownBy(
+                        () ->
+                                catalog.createTable(
+                                        identifier("t_create"),
+                                        csvTableSchema(
+                                                Collections.singletonMap(
+                                                        CsvOptions.SKIP_EMPTY_LINES.key(), "true")),
+                                        false))
+                .satisfies(anyCauseMatches(IllegalArgumentException.class, ERROR));
+    }
+
+    @Test
+    void testRejectedWhenAlteringFileStoreTable() throws Exception {
+        Identifier identifier = identifier("t_alter");
+        catalog.createTable(identifier, csvTableSchema(Collections.emptyMap()), false);
+
+        assertThatThrownBy(
+                        () ->
+                                catalog.alterTable(
+                                        identifier,
+                                        SchemaChange.setOption(
+                                                CsvOptions.SKIP_EMPTY_LINES.key(), "true"),
+                                        false))
+                .satisfies(anyCauseMatches(IllegalArgumentException.class, ERROR));
+    }
+
+    @Test
+    void testRejectedAsDynamicOptionOfFileStoreTable() throws Exception {
+        Identifier identifier = identifier("t_dynamic");
+        catalog.createTable(identifier, csvTableSchema(Collections.emptyMap()), false);
+        Table table = catalog.getTable(identifier);
+        // A single NULL column is written as an empty line, so this is the row that would be
+        // skipped while the file metadata still counts it.
+        write(table, GenericRow.of((Object) null), GenericRow.of(1));
+
+        assertThatThrownBy(
+                        () ->
+                                table.copy(
+                                        Collections.singletonMap(
+                                                CsvOptions.SKIP_EMPTY_LINES.key(), "true")))
+                .satisfies(anyCauseMatches(IllegalArgumentException.class, ERROR));
+
+        // Without the option, the scan and the row count in the metadata agree.
+        assertThat(read(table)).hasSize(2);
+        long metadataRowCount = 0;
+        for (Split split : table.newReadBuilder().newScan().plan().splits()) {
+            metadataRowCount += split.mergedRowCount().getAsLong();
+        }
+        assertThat(metadataRowCount).isEqualTo(2);
+    }
+
+    @Test
+    void testFormatTableSkipsEmptyLinesWithoutRowCount() throws Exception {
+        java.nio.file.Path location = tempPath.resolve("format_table");
+        Files.createDirectories(location);
+        Files.write(location.resolve("data.csv"), "1\n\n \n2\n".getBytes(StandardCharsets.UTF_8));
+
+        Map<String, String> options = new HashMap<>();
+        options.put(CoreOptions.FILE_FORMAT.key(), "csv");
+        options.put(CsvOptions.SKIP_EMPTY_LINES.key(), "true");
+        FormatTable table =
+                FormatTable.builder()
+                        .fileIO(LocalFileIO.create())
+                        .identifier(identifier("t_format"))
+                        .rowType(RowType.of(DataTypes.INT()))
+                        .partitionKeys(Collections.emptyList())
+                        .location(location.toString())
+                        .format(FormatTable.Format.CSV)
+                        .options(options)
+                        .build();
+
+        TableScan.Plan plan = table.newReadBuilder().newScan().plan();
+        // Format table splits carry no row count, so COUNT(*) can never be answered from
+        // metadata and always scans the rows, which skips the empty lines.
+        for (Split split : plan.splits()) {
+            assertThat(split.mergedRowCount()).isEmpty();
+        }
+        List<Integer> ids = new ArrayList<>();
+        try (RecordReader<InternalRow> reader =
+                table.newReadBuilder().newRead().createReader(plan)) {
+            reader.forEachRemaining(row -> ids.add(row.getInt(0)));
+        }
+        assertThat(ids).containsExactly(1, 2);
+    }
+
+    private static Schema csvTableSchema(Map<String, String> extraOptions) {
+        Schema.Builder builder =
+                Schema.newBuilder()
+                        .column("v", DataTypes.INT())
+                        .option(CoreOptions.FILE_FORMAT.key(), "csv")
+                        .option(CoreOptions.FILE_COMPRESSION.key(), "gzip")
+                        .option(CoreOptions.BUCKET.key(), "-1");
+        extraOptions.forEach(builder::option);
+        return builder.build();
+    }
+}

paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFileReader.java

@@ -32,6 +32,7 @@
 public class CsvFileReader extends AbstractTextFileReader {
 
     private final boolean includeHeader;
+    private final boolean skipEmptyLines;
     private final CsvParser csvParser;
 
     private boolean headerSkipped = false;
@@ -47,6 +48,7 @@ public CsvFileReader(
             throws IOException {
         super(fileIO, filePath, projectedRowType, options.lineDelimiter(), offset, length);
         this.includeHeader = options.includeHeader();
+        this.skipEmptyLines = options.skipEmptyLines();
         this.csvParser =
                 new CsvParser(
                         rowReadType,
@@ -55,7 +57,13 @@ public CsvFileReader(
     }
 
     @Override
+    @Nullable
     protected InternalRow parseLine(String line) {
+        // Same test as Spark's CSV source (String#trim drops chars <= U+0020). Returning null
+        // makes the text reader move on to the next line.
+        if (skipEmptyLines && line.trim().isEmpty()) {
+            return null;
+        }
         return csvParser.parse(line);
     }
 

paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java

@@ -72,6 +72,22 @@ public class CsvOptions {
                     .withFallbackKeys("nullvalue")
                     .withDescription("The literal for null values in CSV format");
 
+    public static final ConfigOption<Boolean> SKIP_EMPTY_LINES =
+            ConfigOptions.key("csv.skip-empty-lines")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to skip empty lines when reading, the same way as Spark's CSV "
+                                    + "source. Only supported for format tables. A line is empty if "
+                                    + "nothing is left after String#trim, which removes all "
+                                    + "characters up to U+0020, including spaces, tabs and other "
+                                    + "control characters. By default such a line is read as a row "
+                                    + "of NULLs. Note that Paimon itself writes a row whose values "
+                                    + "are all NULL as an empty line in a single-column table, or as "
+                                    + "a line of delimiters only, which can also be empty after "
+                                    + "trimming (e.g. a tab). Such rows are skipped when this option "
+                                    + "is enabled.");
+
     public static final ConfigOption<Mode> MODE =
             ConfigOptions.key("csv.mode")
                     .enumType(Mode.class)
@@ -87,6 +103,7 @@ public class CsvOptions {
     private final String quoteCharacter;
     private final String escapeCharacter;
     private final Mode mode;
+    private final boolean skipEmptyLines;
 
     public CsvOptions(Options options) {
         this.fieldDelimiter = singleCharacter(options, FIELD_DELIMITER);
@@ -98,6 +115,11 @@ public CsvOptions(Options options) {
         this.quoteCharacter = singleCharacter(options, QUOTE_CHARACTER);
         this.escapeCharacter = singleCharacter(options, ESCAPE_CHARACTER);
         this.mode = options.get(MODE);
+        this.skipEmptyLines = options.get(SKIP_EMPTY_LINES);
+    }
+
+    public boolean skipEmptyLines() {
+        return skipEmptyLines;
     }
 
     /**