[format] Skip the CSV header only in the split that contains it

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;
         }
     }

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<InternalRow> 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<InternalRow> firstSplit = readSplit(format, rowType, testFile, 0, splitPoint);
+            List<InternalRow> 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<InternalRow> 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<InternalRow> read(
         }
     }
 
+    private List<InternalRow> readSplit(
+            FileFormat format, RowType rowType, Path testFile, long offset, long length)
+            throws IOException {
+        try (RecordReader<InternalRow> reader =
+                format.createReaderFactory(rowType, rowType, new ArrayList<>())
+                        .createReader(
+                                new FormatReaderContext(
+                                        fileIO, testFile, fileIO.getFileSize(testFile)),
+                                offset,
+                                length)) {
+
+            InternalRowSerializer serializer = new InternalRowSerializer(rowType);
+            List<InternalRow> 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<InternalRow> 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<InternalRow> 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;
     }
 }