diff --git a/docs/docs/concepts/spec/fileformat.md b/docs/docs/concepts/spec/fileformat.md index 18c1bb59d969..4653f6e48f1d 100644 --- a/docs/docs/concepts/spec/fileformat.md +++ b/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'`: | Paimon CSV format uses [jackson databind API](https://github.com/FasterXML/jackson-databind) to parse and generate CSV string. diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java index c98600f7eefd..e8480d821ac7 100644 --- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java +++ b/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 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 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. diff --git a/paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/CsvSkipEmptyLinesTableTest.java new file mode 100644 index 000000000000..0a11cb8e5b06 --- /dev/null +++ b/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 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 ids = new ArrayList<>(); + try (RecordReader reader = + table.newReadBuilder().newRead().createReader(plan)) { + reader.forEachRemaining(row -> ids.add(row.getInt(0))); + } + assertThat(ids).containsExactly(1, 2); + } + + private static Schema csvTableSchema(Map 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(); + } +} diff --git a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFileReader.java b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFileReader.java index ee11bd9e441d..69288d050be7 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvFileReader.java +++ b/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); } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java b/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java index 8c923ac0f2f0..6e1d821323dc 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/csv/CsvOptions.java +++ b/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 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 = 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; } /** diff --git a/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvSkipEmptyLinesTest.java b/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvSkipEmptyLinesTest.java new file mode 100644 index 000000000000..8ba345e4e8d4 --- /dev/null +++ b/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvSkipEmptyLinesTest.java @@ -0,0 +1,135 @@ +/* + * 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.format.csv; + +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.FileFormatFactory; +import org.apache.paimon.format.FormatReaderContext; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.List; +import java.util.zip.GZIPOutputStream; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link CsvOptions#SKIP_EMPTY_LINES}. */ +class CsvSkipEmptyLinesTest { + + // Tab-separated, with an empty line, a single-space line and a tab-only line in between. + private static final String CONTENT = "a\t1\n\n \nb\t2\n\t\n\r\nc\t3\n"; + + @TempDir java.nio.file.Path tempDir; + + @Test + void testDefaultReadsBlankLinesAsNullRows() throws IOException { + List rows = read(CONTENT, false, "plain.csv"); + assertThat(rows) + .containsExactly( + "a,1", "null,null", "null,null", "b,2", "null,null", "null,null", "c,3"); + } + + @Test + void testSkipEmptyLines() throws IOException { + List rows = read(CONTENT, true, "plain.csv"); + // Like Spark, a line holding only a tab delimiter trims to empty and is skipped too. + assertThat(rows).containsExactly("a,1", "b,2", "c,3"); + } + + @Test + void testSkipEmptyLinesInGzipFile() throws IOException { + List rows = read(CONTENT, true, "data.csv.gz"); + assertThat(rows).containsExactly("a,1", "b,2", "c,3"); + } + + @Test + void testSkipEmptyLinesWithHeader() throws IOException { + Options options = baseOptions(true); + options.set(CsvOptions.INCLUDE_HEADER, true); + List rows = read("h1\th2\n\na\t1\n \nb\t2\n", options, "header.csv"); + assertThat(rows).containsExactly("a,1", "b,2"); + } + + @Test + void testSkipEmptyLinesWithHiveDefaultDelimiter() throws IOException { + // An all-NULL row written with Hive's default delimiter is a single \u0001, which + // String#trim removes, so it is skipped like in Spark. + Options options = baseOptions(true); + options.set(CsvOptions.FIELD_DELIMITER, "\u0001"); + List rows = read("a\u00011\n\u0001\nb\u00012\n", options, "hive.csv"); + assertThat(rows).containsExactly("a,1", "b,2"); + } + + private static Options baseOptions(boolean skipEmptyLines) { + Options options = new Options(); + options.set(CsvOptions.FIELD_DELIMITER, "\t"); + options.set(CsvOptions.SKIP_EMPTY_LINES, skipEmptyLines); + return options; + } + + private List read(String content, boolean skipEmptyLines, String fileName) + throws IOException { + return read(content, baseOptions(skipEmptyLines), fileName); + } + + private List read(String content, Options options, String fileName) throws IOException { + java.nio.file.Path file = tempDir.resolve(fileName); + try (OutputStream out = + fileName.endsWith(".gz") + ? new GZIPOutputStream(Files.newOutputStream(file)) + : Files.newOutputStream(file)) { + out.write(content.getBytes(StandardCharsets.UTF_8)); + } + + CsvFileFormat format = + new CsvFileFormat(new FileFormatFactory.FormatContext(options, 1024, 1024)); + RowType rowType = RowType.of(DataTypes.STRING(), DataTypes.STRING()); + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(file.toString()); + + List rows = new ArrayList<>(); + try (RecordReader reader = + format.createReaderFactory(rowType, rowType, new ArrayList<>()) + .createReader( + new FormatReaderContext( + fileIO, path, fileIO.getFileSize(path), null, null))) { + reader.forEachRemaining( + row -> + rows.add( + (row.isNullAt(0) ? "null" : row.getString(0).toString()) + + "," + + (row.isNullAt(1) + ? "null" + : row.getString(1).toString()))); + } + return rows; + } +}