yihua commented on code in PR #20273:
URL: https://github.com/apache/hudi/pull/20273#discussion_r4264765992
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -188,10 +188,12 @@ public QueryContext analyze() {
}
}
if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.get(0));
+ archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.stream().min(String::compareTo).get());
archivedInstants =
archivedReadTimeline.getInstantsAsStream().filter(instant ->
instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
}
- List<String> instants = Stream.concat(archivedInstants.stream(),
activeInstants.stream()).map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ // archived and active instants can interleave (see
#splitInstantByActiveness), so order them by instant time.
+ List<String> instants = Stream.concat(archivedInstants.stream(),
activeInstants.stream()).map(HoodieInstant::requestedTime)
+ .sorted().collect(Collectors.toList());
Review Comment:
Confirmed this one. With C1 savepointed and C2–C6 archived past it,
`getMaxCompletionTime()` returns C1's completion time, so a bounded Flink read
caps `getLatestMergedFileSlicesBeforeOrOn` at C1 and drops the C2–C6 file
slices; the streaming read picks them up on the next poll since the issued
offset is C1's completion time. `getFilesFromMetadata` also walks the merged
metadata list in reverse assuming ascending order, so an appended C1 can shadow
a newer version of a file group it shares with a hole commit.
`HoodieIncrSource` reads all six commits but checkpoints at C1's completion
time, so it reprocesses C2–C6 in the next batch. The Spark relations use
neither, so they are fine.
All of this needs the savepoint hole, which master already drops entirely,
so I'm fine landing this PR as is and taking it as a follow-up:
`getMaxCompletionTime()` can fall back to the active timeline's latest
completion time whenever an archived instant completed after the newest active
one (same reasoning as the all-archived branch), and `getInstants()` / the
Flink merge can sort by instant time. @ad1happy2go could you open a tracking
issue for that?
--
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]
