zhongyujiang opened a new pull request, #16929: URL: https://github.com/apache/iceberg/pull/16929
## Summary `DataIterator.seek()` overwrites `fileOffset` after `updateCurrentIterator()` has correctly advanced it past files skipped by residual filters. This causes checkpoint to record a wrong file offset, leading to duplicate consumption or `IllegalStateException` on recovery. This is a port of #10567 (Flink 1.19) to Flink 2.1. ## The Bug In `seek(startingFileOffset, startingRecordOffset)`, the call to `updateCurrentIterator()` may advance `fileOffset` beyond `startingFileOffset` when head files are skipped (e.g., their row-group statistics allow the residual filter to eliminate all rows). However, `seek()` then unconditionally executes: ```java fileOffset = startingFileOffset; recordOffset = startingRecordOffset; ``` This overwrites the correct value with a stale one. For example, if `seek(0, 0)` is called and file 0 is skipped by the residual filter, `updateCurrentIterator()` correctly sets `fileOffset = 1` (pointing to file 1). But the assignment resets it to 0. **Consequence:** Checkpoint records `fileOffset = 0` instead of `1`. On recovery: - If file 1 has fewer records than `recordOffset`, throws `IllegalStateException: Invalid starting record offset` - Otherwise, reads from the wrong file → duplicate data ## The Fix 1. Increment `fileOffset` in the skip-files loop (so it tracks position as files are skipped) 2. Remove the incorrect `fileOffset = startingFileOffset` / `recordOffset = startingRecordOffset` assignments at the end of `seek()` — these values are already correct through incremental updates `recordOffset` is also naturally correct after the skip-records loop because `next()` increments it on each call, and the loop is guarded by `currentFileHasNext()` so no file-switch (which would reset `recordOffset`) can occur mid-skip. ## Tests - `testDataIteratorWithResidualFilter`: validates that `fileOffset` matches the actual file being read when a residual filter is applied - `testInitializationWithHeadFilesSkipped`: directly asserts that `fileOffset = 1` after `seek(0, 0)` when file 0 is skipped by residual filter — this fails without the fix cc @pvary @stevenzwu Could you please review? This is a resubmission of #10567 ported to the latest Flink version (2.1), with an improved description. The underlying bug and fix are identical. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
