cshuo commented on code in PR #19553:
URL: https://github.com/apache/hudi/pull/19553#discussion_r3877815023
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -167,51 +172,145 @@ public QueryContext analyze() {
return QueryContext.EMPTY;
}
HoodieTimeline filteredTimeline = getFilteredTimeline(this.metaClient);
- List<String> instantTimeList =
completionTimeQueryView.getInstantTimes(filteredTimeline, startCompletionTime,
endCompletionTime, rangeType);
- if (instantTimeList.isEmpty()) {
- // no instants completed within the give time range, returns early.
- return QueryContext.EMPTY;
- }
- // get hoodie instants
- Pair<List<String>, List<String>> splitInstantTime =
splitInstantByActiveness(instantTimeList, completionTimeQueryView);
- Set<String> instantTimeSet = new HashSet<>(instantTimeList);
- List<String> archivedInstantTimes = splitInstantTime.getLeft();
- List<String> activeInstantTimes = splitInstantTime.getRight();
- List<HoodieInstant> archivedInstants = new ArrayList<>();
- List<HoodieInstant> activeInstants = new ArrayList<>();
- HoodieTimeline archivedReadTimeline = null;
- if (!activeInstantTimes.isEmpty()) {
- activeInstants = filteredTimeline.getInstantsAsStream().filter(instant
->
instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
- if (limit > 0 && limit < activeInstants.size()) {
- // streaming read speed limit, limits the maximum number of commits
allowed to read for each run
- activeInstants = activeInstants.subList(0, limit);
- }
- }
- if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.get(0));
- 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());
- if (instants.isEmpty()) {
- // no instants completed within the give time range, returns early.
- return QueryContext.EMPTY;
- }
- if (startCompletionTime.isEmpty() && endCompletionTime.isPresent()) {
- instants = Collections.singletonList(instants.get(instants.size() -
1));
+ if
(TimelineLayoutVersion.LAYOUT_VERSION_1.equals(metaClient.getTimelineLayoutVersion()))
{
+ return analyzeForV1Timeline(filteredTimeline, completionTimeQueryView);
}
- String lastInstant = instants.get(instants.size() - 1);
- // null => if starting from earliest, if no startCompletionTime is
specified, start from the latest instant like usual streaming read semantics.
- // if startCompletionTime is neither, then use the earliest instant as
the start instant.
- String startInstant =
START_COMMIT_EARLIEST.equalsIgnoreCase(startCompletionTime.orElse(null)) ? null
:
- startCompletionTime.isEmpty() ? lastInstant : instants.get(0);
- String endInstant = endCompletionTime.isEmpty() ? null : lastInstant;
- return QueryContext.create(startInstant, endInstant, instants,
archivedInstants, activeInstants, filteredTimeline, archivedReadTimeline);
+ return analyzeForV2Timeline(filteredTimeline, completionTimeQueryView);
} catch (Exception ex) {
log.error("Got exception when generating incremental query info", ex);
throw new HoodieException(ex);
}
}
+ private QueryContext analyzeForV1Timeline(
+ HoodieTimeline filteredTimeline,
+ CompletionTimeQueryView completionTimeQueryView) {
+ final boolean startFromEarliest =
START_COMMIT_EARLIEST.equalsIgnoreCase(startCompletionTime.orElse(null));
+ final Option<String> effectiveStart = startFromEarliest ? Option.empty() :
startCompletionTime;
+ final HoodieTimeline completedTimeline =
filteredTimeline.filterCompletedInstants();
+
+ if (effectiveStart.isEmpty() && endCompletionTime.isEmpty()) {
+ // (_, _) reads the latest completed instant while [earliest, +INF] is a
snapshot read.
+ List<HoodieInstant> activeInstants = completedTimeline.lastInstant()
+ .map(Collections::singletonList)
+ .orElse(Collections.emptyList());
+ return createQueryContext(Collections.emptyList(), activeInstants,
filteredTimeline, null);
+ }
+
+ List<HoodieInstant> archivedInstants = Collections.emptyList();
+ List<HoodieInstant> activeInstants;
+ HoodieTimeline archivedReadTimeline = null;
+
+ if (startCompletionTime.isEmpty() && endCompletionTime.isPresent()) {
+ // (_, end] returns the last eligible instant at or before end. Check
the filtered active
+ // timeline first and only load the archive when there is no active
match.
+ activeInstants = getLastInstantAtOrBefore(completedTimeline,
endCompletionTime.get());
+ if (activeInstants.isEmpty()) {
+ archivedReadTimeline = getArchivedReadTimeline(metaClient, "");
+ archivedInstants = getLastInstantAtOrBefore(archivedReadTimeline,
endCompletionTime.get());
+ }
+ } else {
+ InstantRange instantRange = InstantRange.builder()
+ .rangeType(rangeType)
+ .startInstant(effectiveStart.orElse(null))
+ .endInstant(endCompletionTime.orElse(null))
+ .nullableBoundary(true)
+ .build();
+ activeInstants = getInstantsInRange(completedTimeline, instantRange);
+
+ boolean needsArchivedInstants = startFromEarliest
+ || (effectiveStart.isPresent() &&
completionTimeQueryView.isArchived(effectiveStart.get()));
+ if (needsArchivedInstants) {
+ archivedReadTimeline = getArchivedReadTimeline(metaClient,
effectiveStart.orElse(""));
+ archivedInstants = getInstantsInRange(archivedReadTimeline,
instantRange);
+ }
+ }
+
+ return createQueryContext(
+ archivedInstants, activeInstants, filteredTimeline,
archivedReadTimeline);
+ }
+
+ private QueryContext analyzeForV2Timeline(
+ HoodieTimeline filteredTimeline,
+ CompletionTimeQueryView completionTimeQueryView) {
+ List<String> instantTimeList = completionTimeQueryView.getInstantTimes(
+ filteredTimeline, startCompletionTime, endCompletionTime, rangeType);
+ if (instantTimeList.isEmpty()) {
+ // no instants completed within the given time range, returns early.
+ return QueryContext.EMPTY;
+ }
+
+ Pair<List<String>, List<String>> splitInstantTime =
splitInstantByActiveness(
+ instantTimeList, completionTimeQueryView);
+ Set<String> instantTimeSet = new HashSet<>(instantTimeList);
+ List<String> archivedInstantTimes = splitInstantTime.getLeft();
+ List<String> activeInstantTimes = splitInstantTime.getRight();
+ List<HoodieInstant> archivedInstants = new ArrayList<>();
+ List<HoodieInstant> activeInstants = new ArrayList<>();
+ HoodieTimeline archivedReadTimeline = null;
+ if (!activeInstantTimes.isEmpty()) {
+ activeInstants = filteredTimeline.getInstantsAsStream()
+ .filter(instant -> instantTimeSet.contains(instant.requestedTime()))
+ .collect(Collectors.toList());
+ }
+ if (!archivedInstantTimes.isEmpty()) {
+ archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.get(0));
+ archivedInstants = archivedReadTimeline.getInstantsAsStream()
+ .filter(instant -> instantTimeSet.contains(instant.requestedTime()))
+ .collect(Collectors.toList());
+ }
+ return createQueryContext(
+ archivedInstants, activeInstants, filteredTimeline,
archivedReadTimeline);
+ }
+
+ private QueryContext createQueryContext(
+ List<HoodieInstant> archivedInstants,
+ List<HoodieInstant> activeInstants,
+ HoodieTimeline filteredTimeline,
+ @Nullable HoodieTimeline archivedReadTimeline) {
+ if (limit > 0 && limit < activeInstants.size()) {
+ // streaming read speed limit, limits the maximum number of active
commits allowed per run
+ activeInstants = activeInstants.subList(0, limit);
+ }
+ List<String> instants = Stream.concat(archivedInstants.stream(),
activeInstants.stream())
Review Comment:
This concatenation does not guarantee global requested-time ordering. With
`hoodie.archive.beyond.savepoint=true`, an old savepointed commit may remain
active while later commits are archived. For example, archived `[102, 103,
104]` and active `[101]` produces `[102, 103, 104, 101]`. Since `lastInstant`
is taken from the final list element, the resulting end boundary regresses to
`101`, and downstream readers can omit `102`–`104`.
Please globally deduplicate and sort the combined requested-time list before
deriving `lastInstant`. The previous V1 candidate path explicitly applied both
`distinct()` and `sorted()`, so preserving those properties would also handle
overlap during concurrent archival.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -167,51 +172,145 @@ public QueryContext analyze() {
return QueryContext.EMPTY;
}
HoodieTimeline filteredTimeline = getFilteredTimeline(this.metaClient);
- List<String> instantTimeList =
completionTimeQueryView.getInstantTimes(filteredTimeline, startCompletionTime,
endCompletionTime, rangeType);
- if (instantTimeList.isEmpty()) {
- // no instants completed within the give time range, returns early.
- return QueryContext.EMPTY;
- }
- // get hoodie instants
- Pair<List<String>, List<String>> splitInstantTime =
splitInstantByActiveness(instantTimeList, completionTimeQueryView);
- Set<String> instantTimeSet = new HashSet<>(instantTimeList);
- List<String> archivedInstantTimes = splitInstantTime.getLeft();
- List<String> activeInstantTimes = splitInstantTime.getRight();
- List<HoodieInstant> archivedInstants = new ArrayList<>();
- List<HoodieInstant> activeInstants = new ArrayList<>();
- HoodieTimeline archivedReadTimeline = null;
- if (!activeInstantTimes.isEmpty()) {
- activeInstants = filteredTimeline.getInstantsAsStream().filter(instant
->
instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
- if (limit > 0 && limit < activeInstants.size()) {
- // streaming read speed limit, limits the maximum number of commits
allowed to read for each run
- activeInstants = activeInstants.subList(0, limit);
- }
- }
- if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.get(0));
- 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());
- if (instants.isEmpty()) {
- // no instants completed within the give time range, returns early.
- return QueryContext.EMPTY;
- }
- if (startCompletionTime.isEmpty() && endCompletionTime.isPresent()) {
- instants = Collections.singletonList(instants.get(instants.size() -
1));
+ if
(TimelineLayoutVersion.LAYOUT_VERSION_1.equals(metaClient.getTimelineLayoutVersion()))
{
+ return analyzeForV1Timeline(filteredTimeline, completionTimeQueryView);
}
- String lastInstant = instants.get(instants.size() - 1);
- // null => if starting from earliest, if no startCompletionTime is
specified, start from the latest instant like usual streaming read semantics.
- // if startCompletionTime is neither, then use the earliest instant as
the start instant.
- String startInstant =
START_COMMIT_EARLIEST.equalsIgnoreCase(startCompletionTime.orElse(null)) ? null
:
- startCompletionTime.isEmpty() ? lastInstant : instants.get(0);
- String endInstant = endCompletionTime.isEmpty() ? null : lastInstant;
- return QueryContext.create(startInstant, endInstant, instants,
archivedInstants, activeInstants, filteredTimeline, archivedReadTimeline);
+ return analyzeForV2Timeline(filteredTimeline, completionTimeQueryView);
} catch (Exception ex) {
log.error("Got exception when generating incremental query info", ex);
throw new HoodieException(ex);
}
}
+ private QueryContext analyzeForV1Timeline(
+ HoodieTimeline filteredTimeline,
+ CompletionTimeQueryView completionTimeQueryView) {
+ final boolean startFromEarliest =
START_COMMIT_EARLIEST.equalsIgnoreCase(startCompletionTime.orElse(null));
+ final Option<String> effectiveStart = startFromEarliest ? Option.empty() :
startCompletionTime;
+ final HoodieTimeline completedTimeline =
filteredTimeline.filterCompletedInstants();
+
+ if (effectiveStart.isEmpty() && endCompletionTime.isEmpty()) {
+ // (_, _) reads the latest completed instant while [earliest, +INF] is a
snapshot read.
+ List<HoodieInstant> activeInstants = completedTimeline.lastInstant()
+ .map(Collections::singletonList)
+ .orElse(Collections.emptyList());
+ return createQueryContext(Collections.emptyList(), activeInstants,
filteredTimeline, null);
+ }
+
+ List<HoodieInstant> archivedInstants = Collections.emptyList();
+ List<HoodieInstant> activeInstants;
+ HoodieTimeline archivedReadTimeline = null;
+
+ if (startCompletionTime.isEmpty() && endCompletionTime.isPresent()) {
+ // (_, end] returns the last eligible instant at or before end. Check
the filtered active
+ // timeline first and only load the archive when there is no active
match.
+ activeInstants = getLastInstantAtOrBefore(completedTimeline,
endCompletionTime.get());
Review Comment:
This shortcut assumes that any eligible active instant is newer than every
archived instant, which is not guaranteed with
`hoodie.archive.beyond.savepoint=true`. A savepointed commit can remain active
while later commits are archived; for example, active `[101, 105]` and archived
`[102, 103, 104]`. For an end-only query ending at `104`, this returns active
instant `101` without checking the archive, omitting the actual latest eligible
instant `104`.
Please only use the active-only shortcut when the selected active instant is
on or after the archive boundary; otherwise compare the best active and
archived candidates. A regression test covering this savepoint-hole scenario
would also be helpful.
--
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]
