From 7bb38572da076a4c01d7f7024223be5026355e48 Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Thu, 6 Aug 2026 15:02:00 +0900 Subject: [PATCH] [format] Skip the CSV header only in the split that contains it CsvFileReader#setupReading skipped a line on every split, but the header only exists at byte 0. Splits with a non-zero offset therefore dropped their own first data row, so a header-bearing CSV larger than source.split.target-size silently lost one row per extra split. Generated-by: Claude Code --- .../paimon/format/csv/CsvFileReader.java | 10 ++- .../paimon/format/csv/CsvFileFormatTest.java | 68 ++++++++++++++++++- 2 files changed, 74 insertions(+), 4 deletions(-) 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 5a8f9ac81fd7..ee11bd9e441d 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 @@ -61,9 +61,15 @@ protected InternalRow parseLine(String line) { @Override protected void setupReading() throws IOException { - // Skip header if needed + // Skip header if needed. The header only lives at byte 0, so only the split starting there + // has one to skip. A split with a non-zero offset must not drop a line here: the record + // straddling its start boundary was already discarded by StandardLineReader#skipFirstLine + // and belongs to the previous split, which reads it in full. Dropping another line would + // silently lose the first data row of this split. if (includeHeader && !headerSkipped) { - readLine(); + if (offset == 0) { + readLine(); + } headerSkipped = true; } } diff --git a/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java index 8e3b223a8200..10982d448ae1 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/csv/CsvFileFormatTest.java @@ -391,6 +391,44 @@ public void testCsvIncludeHeaderWriteRead() throws IOException { } } + @Test + public void testHeaderSkippedOnlyInFirstSplit() throws IOException { + RowType rowType = DataTypes.ROW(DataTypes.INT().notNull(), DataTypes.STRING()); + + List testData = new ArrayList<>(); + for (int i = 0; i < 50; i++) { + testData.add(GenericRow.of(i, BinaryString.fromString("name" + i))); + } + + for (boolean includeHeader : new boolean[] {true, false}) { + Options options = new Options(); + options.set(CsvOptions.INCLUDE_HEADER, includeHeader); + FileFormat format = + new CsvFileFormatFactory().create(new FormatContext(options, 1024, 1024)); + Path testFile = write(format, rowType, testData, "test_header_split_" + includeHeader); + + // a single split still skips the header exactly once + assertThat(read(format, rowType, rowType, testFile)).hasSize(testData.size()); + + long fileSize = fileIO.getFileSize(testFile); + long splitPoint = fileSize / 2; + List firstSplit = readSplit(format, rowType, testFile, 0, splitPoint); + List secondSplit = + readSplit(format, rowType, testFile, splitPoint, fileSize - splitPoint); + + // both splits carry data, and together they lose no row and duplicate none + assertThat(firstSplit).isNotEmpty(); + assertThat(secondSplit).isNotEmpty(); + List allRows = new ArrayList<>(firstSplit); + allRows.addAll(secondSplit); + assertThat(allRows).hasSize(testData.size()); + for (int i = 0; i < testData.size(); i++) { + assertThat(allRows.get(i).getInt(0)).isEqualTo(i); + assertThat(allRows.get(i).getString(1).toString()).isEqualTo("name" + i); + } + } + } + @Test public void testCsvNullLiteralWriteRead() throws IOException { RowType rowType = @@ -621,6 +659,24 @@ private List read( } } + private List readSplit( + FileFormat format, RowType rowType, Path testFile, long offset, long length) + throws IOException { + try (RecordReader reader = + format.createReaderFactory(rowType, rowType, new ArrayList<>()) + .createReader( + new FormatReaderContext( + fileIO, testFile, fileIO.getFileSize(testFile)), + offset, + length)) { + + InternalRowSerializer serializer = new InternalRowSerializer(rowType); + List result = new ArrayList<>(); + reader.forEachRemaining(row -> result.add(serializer.copy(row))); + return result; + } + } + @Override protected RowType rowTypeForFullTypesTest() { RowType.Builder builder = @@ -749,15 +805,23 @@ private List writeThenRead( throws IOException { FileFormat format = new CsvFileFormatFactory().create(new FormatContext(options, 1024, 1024)); + Path testFile = write(format, fullRowType, testData, testPrefix); + return read(format, fullRowType, rowType, testFile); + } + + /** Writes the given data to a new CSV file and returns its path. */ + private Path write( + FileFormat format, RowType rowType, List testData, String testPrefix) + throws IOException { Path testFile = new Path(parent, testPrefix + "_" + UUID.randomUUID() + ".csv"); - FormatWriterFactory writerFactory = format.createWriterFactory(fullRowType); + FormatWriterFactory writerFactory = format.createWriterFactory(rowType); try (PositionOutputStream out = fileIO.newOutputStream(testFile, false); FormatWriter writer = writerFactory.create(out, "none")) { for (InternalRow row : testData) { writer.addElement(row); } } - return read(format, fullRowType, rowType, testFile); + return testFile; } }