Conversation
| // filter log files by instant range. | ||
| if (logFiles != null && readerContext.getInstantRange().isPresent()) { | ||
| InstantRange instantRange = readerContext.getInstantRange().get(); | ||
| logFiles = logFiles.filter(logFile -> instantRange.isInRange(logFile.getDeltaCommitTime())); |
There was a problem hiding this comment.
logFiles.filter(logFile -> instantRange.isInRange(logFile.getDeltaCommitTime())); this already works, the logFile.getDeltaCommitTime is the delta comit time of the instant.
There was a problem hiding this comment.
The existing filter works when InstantRange is present, but Spark time-travel queries leave it empty and pass the as-of timestamp through latestCommitTime. A selected slice can still contain later logs, so we need this cutoff.
Replacing latestCommitTime with InstantRange in the reader context could be handled in a follow-up refactoring PR.
There was a problem hiding this comment.
🤖 For what it's worth, this matches what the legacy log path already does independently of InstantRange: BaseHoodieLogRecordReader (around L296) skips any data/delete block whose INSTANT_TIME header is greater than readerContext.getLatestCommitTime(), and only applies InstantRange on top of that. The Spark time-travel path sets latestCommitTime without an InstantRange, so the LSM reader had no upper bound at all — this change just brings it to parity at file granularity.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19984 +/- ##
============================================
+ Coverage 80.33% 80.39% +0.05%
- Complexity 34752 34941 +189
============================================
Files 2543 2544 +1
Lines 142478 143038 +560
Branches 17308 17462 +154
============================================
+ Hits 114464 114995 +531
- Misses 20119 20130 +11
- Partials 7895 7913 +18
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR makes HoodieLsmFileGroupReader drop native data/delete log files whose delta commit time is newer than latestCommitTime, mirroring the block-level upper-bound check the classic log scanner applies, so time-travel and bounded reads no longer see future updates/deletes. I traced the callers (Spark file format / MOR RDD, Flink FormatUtils incl. streaming/CDC/bootstrap/clustering splits, MDT reader, and the LSM merge handle for upsert/compaction) and the latestCommitTime each passes is already the intended upper bound for the equivalent classic path, so the new unconditional filter is consistent with existing semantics. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
| // filter log files by instant range. | ||
| if (logFiles != null && readerContext.getInstantRange().isPresent()) { | ||
| InstantRange instantRange = readerContext.getInstantRange().get(); | ||
| logFiles = logFiles.filter(logFile -> instantRange.isInRange(logFile.getDeltaCommitTime())); |
There was a problem hiding this comment.
🤖 For what it's worth, this matches what the legacy log path already does independently of InstantRange: BaseHoodieLogRecordReader (around L296) skips any data/delete block whose INSTANT_TIME header is greater than readerContext.getLatestCommitTime(), and only applies InstantRange on top of that. The Spark time-travel path sets latestCommitTime without an InstantRange, so the LSM reader had no upper bound at all — this change just brings it to parity at file granularity.
5 participants
Describe the issue this Pull Request addresses
Closes #19983.
HoodieLsmFileGroupReadercan read native data/delete logs newer thanlatestCommitTime, causing time-travel reads to apply future updates or deletions. Selecting a file slice by its base instant does not exclude later logs within that slice. For example, a delete at004must not remove a record when reading as of002.Summary and Changelog
deltaCommitTime <= latestCommitTimebefore constructing the input split, then apply the existing optionalInstantRange.Impact
LSM reads exclude updates and deletes after the query instant. Excluded log files are not opened. No public API, storage format, or configuration changes.
Risk Level
Low. The change applies the existing classic scanner's upper-bound semantics at native-log file level. Regression tests reproduce the failure before the fix. All 24 tests across
TestHoodieLsmFileGroupReader,TestLsmFileGroupRecordIterator,TestLsmFileIterators,TestSpillableLsmRecordIterator, andTestLsmReaderUtilspass after the fix. Maven Checkstyle andgit diff --checkalso pass.Documentation Update
None.
Contributor's checklist