This is an automated email from the ASF dual-hosted git repository.
leonardBang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ad41a62bf9b [FLINK-20746][csv] Add csv.ignore-first-line for
filesystem CSV connector
ad41a62bf9b is described below
commit ad41a62bf9b4223dfd23eb335fd4241c5df58a4e
Author: Vishal Kamlapure <[email protected]>
AuthorDate: Tue Sep 15 08:20:36 2026 +0530
[FLINK-20746][csv] Add csv.ignore-first-line for filesystem CSV connector
This closes #28594.
---
.../docs/connectors/table/formats/csv.md | 8 +++
docs/content/docs/connectors/table/formats/csv.md | 8 +++
.../flink/formats/csv/CsvFileFormatFactory.java | 8 ++-
.../apache/flink/formats/csv/CsvFormatOptions.java | 9 +++
.../flink/formats/csv/CsvFormatFactoryTest.java | 81 ++++++++++++++++++++++
5 files changed, 113 insertions(+), 1 deletion(-)
diff --git a/docs/content.zh/docs/connectors/table/formats/csv.md
b/docs/content.zh/docs/connectors/table/formats/csv.md
index 348c7bd043f..846005a6d10 100644
--- a/docs/content.zh/docs/connectors/table/formats/csv.md
+++ b/docs/content.zh/docs/connectors/table/formats/csv.md
@@ -199,6 +199,14 @@ Format 参数
<td>Boolean</td>
<td>可选标志，用于将空字符串值视为 null（默认禁用）。仅影响反序列化。</td>
</tr>
+ <tr>
+ <td><h5>csv.ignore-first-line</h5></td>
+ <td>可选</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>可选标志，用于跳过每个输入 CSV 文件的第一条记录（默认值为 false）。仅支持 filesystem
连接器。仅影响反序列化。</td>
+ </tr>
</tbody>
</table>
diff --git a/docs/content/docs/connectors/table/formats/csv.md
b/docs/content/docs/connectors/table/formats/csv.md
index 7262757459a..f14bf82269c 100644
--- a/docs/content/docs/connectors/table/formats/csv.md
+++ b/docs/content/docs/connectors/table/formats/csv.md
@@ -200,6 +200,14 @@ Format Options
<td>Boolean</td>
<td>Optional flag to treat empty string values as null (disabled by
default). Only affects deserialization.</td>
</tr>
+ <tr>
+ <td><h5>csv.ignore-first-line</h5></td>
+ <td>optional</td>
+ <td>no</td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Optional flag to skip the first CSV record of each input file (false
by default). Only supported by the filesystem connector. Only affects
deserialization.</td>
+ </tr>
</tbody>
</table>
diff --git
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
index e58796bdb9c..d69dd42264e 100644
---
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
+++
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java
@@ -58,6 +58,7 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.Csv
import org.apache.commons.text.StringEscapeUtils;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -69,6 +70,7 @@ import static
org.apache.flink.formats.csv.CsvFormatOptions.EMPTY_STRING_AS_NULL
import static org.apache.flink.formats.csv.CsvFormatOptions.ESCAPE_CHARACTER;
import static
org.apache.flink.formats.csv.CsvFormatOptions.FAIL_ON_MISSING_COLUMNS;
import static org.apache.flink.formats.csv.CsvFormatOptions.FIELD_DELIMITER;
+import static org.apache.flink.formats.csv.CsvFormatOptions.IGNORE_FIRST_LINE;
import static
org.apache.flink.formats.csv.CsvFormatOptions.IGNORE_PARSE_ERRORS;
import static
org.apache.flink.formats.csv.CsvFormatOptions.IGNORE_TRAILING_UNMAPPABLE;
import static org.apache.flink.formats.csv.CsvFormatOptions.NULL_LITERAL;
@@ -92,7 +94,9 @@ public class CsvFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
@Override
public Set<ConfigOption<?>> optionalOptions() {
- return CsvCommons.optionalOptions();
+ Set<ConfigOption<?>> options = new
HashSet<>(CsvCommons.optionalOptions());
+ options.add(IGNORE_FIRST_LINE);
+ return options;
}
@Override
@@ -212,6 +216,8 @@ public class CsvFileFormatFactory implements
BulkReaderFormatFactory, BulkWriter
options.getOptional(ALLOW_COMMENTS).ifPresent(csvBuilder::setAllowComments);
+
options.getOptional(IGNORE_FIRST_LINE).ifPresent(csvBuilder::setSkipFirstDataRow);
+
options.getOptional(ARRAY_ELEMENT_DELIMITER)
.ifPresent(csvBuilder::setArrayElementSeparator);
diff --git
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatOptions.java
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatOptions.java
index ee624a48e0b..c5f05122ff8 100644
---
a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatOptions.java
+++
b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatOptions.java
@@ -139,5 +139,14 @@ public class CsvFormatOptions {
+ "(disabled by default). "
+ "Only affects deserialization.");
+ public static final ConfigOption<Boolean> IGNORE_FIRST_LINE =
+ ConfigOptions.key("ignore-first-line")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Optional flag to skip the first CSV record of
each input file "
+ + "(false by default). "
+ + "Only affects deserialization.");
+
private CsvFormatOptions() {}
}
diff --git
a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java
b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java
index fbe1c993c74..ba56f8e9cad 100644
---
a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java
+++
b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java
@@ -23,6 +23,8 @@ import
org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.Utils;
+import org.apache.flink.core.fs.Path;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Column;
@@ -36,11 +38,17 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.TestDynamicTableFactory;
import
org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.DataType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -62,6 +70,15 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link CsvFormatFactory}. */
class CsvFormatFactoryTest {
+ private static final DataType HEADER_ROW_DATA_TYPE =
+ ResolvedSchema.of(
+ Column.physical("id", DataTypes.STRING()),
+ Column.physical("name", DataTypes.STRING()),
+ Column.physical("amount", DataTypes.STRING()))
+ .toPhysicalRowDataType();
+
+ private static final String CSV_WITH_HEADER =
"id,name,amount\n1,Alice,100\n2,Bob,200";
+
@Test
void testSeDeSchema() {
final CsvRowDataDeserializationSchema expectedDeser =
@@ -456,6 +473,43 @@ class CsvFormatFactoryTest {
assertThat(bulkFormat).isNotNull();
}
+ @Test
+ void testIgnoreFirstLineSkipsFirstRecord(@TempDir java.nio.file.Path
tempDir) throws Exception {
+ final File file = tempDir.resolve("with-header.csv").toFile();
+ Files.writeString(file.toPath(), CSV_WITH_HEADER,
StandardCharsets.UTF_8);
+
+ final Configuration formatOptions = new Configuration();
+ formatOptions.set(CsvFormatOptions.IGNORE_FIRST_LINE, true);
+
+ final List<RowData> rows = readBulkFormatRows(formatOptions, file,
HEADER_ROW_DATA_TYPE);
+
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo("1");
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo("Alice");
+ assertThat(rows.get(0).getString(2).toString()).isEqualTo("100");
+ assertThat(rows.get(1).getString(0).toString()).isEqualTo("2");
+ assertThat(rows.get(1).getString(1).toString()).isEqualTo("Bob");
+ assertThat(rows.get(1).getString(2).toString()).isEqualTo("200");
+ }
+
+ @Test
+ void testIgnoreFirstLineDisabledByDefault(@TempDir java.nio.file.Path
tempDir)
+ throws Exception {
+ final File file = tempDir.resolve("with-header.csv").toFile();
+ Files.writeString(file.toPath(), CSV_WITH_HEADER,
StandardCharsets.UTF_8);
+
+ final List<RowData> rows =
+ readBulkFormatRows(new Configuration(), file,
HEADER_ROW_DATA_TYPE);
+
+ assertThat(rows).hasSize(3);
+ assertThat(rows.get(0).getString(0).toString()).isEqualTo("id");
+ assertThat(rows.get(0).getString(1).toString()).isEqualTo("name");
+ assertThat(rows.get(0).getString(2).toString()).isEqualTo("amount");
+ assertThat(rows.get(1).getString(0).toString()).isEqualTo("1");
+ assertThat(rows.get(1).getString(2).toString()).isEqualTo("100");
+ assertThat(rows.get(2).getString(1).toString()).isEqualTo("Bob");
+ }
+
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
@@ -490,6 +544,33 @@ class CsvFormatFactoryTest {
return options;
}
+ private static List<RowData> readBulkFormatRows(
+ Configuration formatOptions, File file, DataType dataType) throws
IOException {
+ final CsvFileFormatFactory.CsvBulkDecodingFormat bulkDecodingFormat =
+ new CsvFileFormatFactory.CsvBulkDecodingFormat(formatOptions);
+ final BulkFormat<RowData, FileSourceSplit> bulkFormat =
+ bulkDecodingFormat.createRuntimeDecoder(
+ ScanRuntimeProviderContext.INSTANCE,
+ dataType,
+ Projection.all(dataType).toNestedIndexes());
+
+ final FileSourceSplit split =
+ new FileSourceSplit(
+ "split-0",
+ new Path(file.toURI()),
+ 0,
+ file.length(),
+ file.lastModified(),
+ file.length());
+
+ final List<RowData> rows = new ArrayList<>();
+ try (BulkFormat.Reader<RowData> reader =
+ bulkFormat.createReader(new Configuration(), split)) {
+ Utils.forEachRemaining(reader, rows::add);
+ }
+ return rows;
+ }
+
private static DeserializationSchema<RowData> createDeserializationSchema(
Map<String, String> options) {
TestDynamicTableFactory.DynamicTableSourceMock sourceMock =
