hudi-agent commented on code in PR #19338:
URL: https://github.com/apache/hudi/pull/19338#discussion_r3728097566
##########
hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java:
##########
@@ -205,7 +217,57 @@ private List<String> getInstantTimes(
Option<String> rangeEnd,
InstantRange.RangeType rangeType,
Function<String, String> earliestInstantTimeFunc) {
- throw new RuntimeException("Incremental query view for timeline version 1
not yet implemented");
+ final boolean startFromEarliest =
START_COMMIT_EARLIEST.equalsIgnoreCase(rangeStart.orElse(null));
+ final Option<String> effectiveStart = startFromEarliest ? Option.empty() :
rangeStart;
+ final HoodieTimeline completedTimeline =
timeline.filterCompletedInstants();
+
+ if (rangeStart.isEmpty() && rangeEnd.isPresent()) {
+ // (_, end]: returns the last instant whose requested time is at or
before 'end'.
+ Option<String> latestActiveInstant =
Option.fromJavaOptional(completedTimeline.getInstantsAsStream()
+ .filter(instant ->
InstantComparison.compareTimestamps(instant.requestedTime(),
LESSER_THAN_OR_EQUALS, rangeEnd.get()))
+ .max(Comparator.comparing(HoodieInstant::requestedTime))
+ .map(HoodieInstant::requestedTime));
+ if (latestActiveInstant.isPresent()) {
+ return Collections.singletonList(latestActiveInstant.get());
+ }
+ // The archived timeline is not filtered by user configs yet, so return
all candidates and
+ // let IncrementalQueryAnalyzer select the last eligible instant after
applying those filters.
+ return getArchivedInstantTimes(Option.empty())
+ .filter(instant -> InstantComparison.compareTimestamps(instant,
LESSER_THAN_OR_EQUALS, rangeEnd.get()))
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ if (effectiveStart.isEmpty() && rangeEnd.isEmpty()) {
+ // (_, _): read the latest snapshot.
+ return completedTimeline.lastInstant().map(instant ->
Collections.singletonList(instant.requestedTime())).orElse(Collections.emptyList());
+ }
+
+ final InstantRange instantRange = InstantRange.builder()
+ .rangeType(rangeType)
+ .startInstant(effectiveStart.orElse(null))
+ .endInstant(rangeEnd.orElse(null))
+ .nullableBoundary(true)
+ .build();
+ Stream<String> candidateInstants =
completedTimeline.getInstantsAsStream().map(HoodieInstant::requestedTime);
+ final boolean needsArchivedInstants = startFromEarliest
+ || (effectiveStart.isPresent() && isArchived(effectiveStart.get()));
+ if (needsArchivedInstants) {
+ candidateInstants =
Stream.concat(getArchivedInstantTimes(effectiveStart), candidateInstants);
+ }
+ return candidateInstants
+ .filter(instantRange::isInRange)
+ .distinct()
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private Stream<String> getArchivedInstantTimes(Option<String> startInstant) {
Review Comment:
🤖 When the query range includes archived instants, the archived timeline
ends up loaded twice per `analyze()` — once here with `useCache=false`, and
again in `IncrementalQueryAnalyzer.getArchivedReadTimeline` (also
`useCache=false`). The `getArchivedTimeline` javadoc flags this as costly, and
the `(_, end]` fallback loads from `""` (the whole archived timeline). Since V2
amortizes this via its in-memory map, is there a way to avoid the second full
load on the catch-up path (e.g. `useCache=true`, or letting the view/analyzer
share one load)?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
