cshuo commented on code in PR #19553:
URL: https://github.com/apache/hudi/pull/19553#discussion_r3891530894
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java:
##########
@@ -167,51 +172,159 @@ 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());
+ if
(TimelineLayoutVersion.LAYOUT_VERSION_1.equals(metaClient.getTimelineLayoutVersion()))
{
+ return analyzeForV1Timeline(filteredTimeline, completionTimeQueryView);
}
- 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));
- }
- 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(
Review Comment:
nit: Could you add some method comments so that other developers can easily
understand the optimizations in `analyzeForV1Timeline` compared with
`analyzeForV2Timeline`?
--
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]
