Flink-1.19: Fix the file offset mismatch when Flink reader first seek…

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/DataIterator.java

@@ -85,6 +85,7 @@ public void seek(int startingFileOffset, long startingRecordOffset) {
         combinedTask);
     for (long i = 0L; i < startingFileOffset; ++i) {
       tasks.next();
+      fileOffset += 1;
     }
 
     updateCurrentIterator();
@@ -99,9 +100,6 @@ public void seek(int startingFileOffset, long startingRecordOffset) {
                 startingRecordOffset, startingFileOffset, combinedTask));
       }
     }
-
-    fileOffset = startingFileOffset;
-    recordOffset = startingRecordOffset;
   }
 
   @Override

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/reader/ReaderUtil.java

@@ -60,6 +60,22 @@ public static FileScanTask createFileTask(
       FileFormat fileFormat,
       FileAppenderFactory<Record> appenderFactory)
       throws IOException {
+
+    return createFileTask(
+        records,
+        file,
+        fileFormat,
+        appenderFactory,
+        ResidualEvaluator.unpartitioned(Expressions.alwaysTrue()));
+  }
+
+  public static FileScanTask createFileTask(
+      List<Record> records,
+      File file,
+      FileFormat fileFormat,
+      FileAppenderFactory<Record> appenderFactory,
+      ResidualEvaluator residuals)
+      throws IOException {
     FileAppender<Record> appender =
         appenderFactory.newAppender(Files.localOutput(file), fileFormat);
     try {
@@ -77,7 +93,6 @@ public static FileScanTask createFileTask(
             .withMetrics(appender.metrics())
             .build();
 
-    ResidualEvaluator residuals = ResidualEvaluator.unpartitioned(Expressions.alwaysTrue());
     return new BaseFileScanTask(
         dataFile,
         null,

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/source/reader/TestArrayPoolDataIteratorBatcherRowData.java

@@ -21,6 +21,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.File;
+import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.List;
@@ -33,13 +34,17 @@
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.data.GenericAppenderFactory;
+import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.RandomGenericData;
 import org.apache.iceberg.data.Record;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.ResidualEvaluator;
 import org.apache.iceberg.flink.FlinkConfigOptions;
 import org.apache.iceberg.flink.TestFixtures;
 import org.apache.iceberg.flink.TestHelpers;
 import org.apache.iceberg.flink.source.DataIterator;
 import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -357,4 +362,46 @@ public void testMultipleFilesWithSeekPosition() throws Exception {
 
     assertThat(recordBatchIterator).isExhausted();
   }
+
+  @Test
+  public void testInitializationWithHeadFilesSkipped() throws IOException {
+    GenericRecord record0 = GenericRecord.create(TestFixtures.SCHEMA);
+    record0.setField("data", "a");
+    record0.setField("id", 1L);
+    record0.setField("dt", "-");
+
+    GenericRecord record1 = GenericRecord.create(TestFixtures.SCHEMA);
+    record1.setField("data", "a");
+    record1.setField("id", 10L);
+    record1.setField("dt", "-");
+
+    ResidualEvaluator residuals = ResidualEvaluator.unpartitioned(Expressions.greaterThan("id", 5));
+
+    FileScanTask fileTask0 =
+        ReaderUtil.createFileTask(
+            ImmutableList.of(record0),
+            File.createTempFile("junit", null, temporaryFolder.toFile()),
+            FILE_FORMAT,
+            appenderFactory,
+            residuals);
+
+    FileScanTask fileTask1 =
+        ReaderUtil.createFileTask(
+            ImmutableList.of(record1),
+            File.createTempFile("junit", null, temporaryFolder.toFile()),
+            FILE_FORMAT,
+            appenderFactory,
+            residuals);
+
+    CombinedScanTask combinedTask = new BaseCombinedScanTask(Arrays.asList(fileTask0, fileTask1));
+
+    DataIterator<RowData> dataIterator = ReaderUtil.createDataIterator(combinedTask);
+
+    // split initialization
+    dataIterator.seek(0, 0);
+
+    assertThat(dataIterator.fileOffset())
+        .as("File offset should be 1 because file 0 should be skipped")
+        .isEqualTo(1);
+  }
 }