You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/10/08 12:07:38 UTC
(hudi) 08/17: feat(common): query LAYOUT_VERSION_1 timelines in CompletionTimeQueryView by using instant time as completion time (#19338)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch release-1.1.2
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 81577d2dcfbbf50d3df8182f7339a7e2d1e6d54a
Author: fhan <aa...@gmail.com>
AuthorDate: Fri Aug 7 09:56:30 2026 +0800
feat(common): query LAYOUT_VERSION_1 timelines in CompletionTimeQueryView by using instant time as completion time (#19338)
* feat(common): query LAYOUT_VERSION_1 timelines in CompletionTimeQueryView by using instant time as completion time
---------
Co-authored-by: fhan <yf...@jd.com>
(cherry picked from commit b88f235174613ef8a0bf980be659f7eb674b461e)
---
.../apache/hudi/common/table/log/InstantRange.java | 4 +-
.../versioning/v1/CompletionTimeQueryViewV1.java | 80 ++++++++-
.../v1/TestCompletionTimeQueryViewV1.java | 189 +++++++++++++++++++++
3 files changed, 262 insertions(+), 11 deletions(-)
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java
index 6d6d29bb40f6..a49cef856d35 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/InstantRange.java
@@ -113,7 +113,7 @@ public abstract class InstantRange implements Serializable {
public OpenClosedRangeNullableBoundary(String startInstant, String endInstant) {
super(startInstant, endInstant);
- ValidationUtils.checkArgument(!startInstant.isEmpty() || !endInstant.isEmpty(),
+ ValidationUtils.checkArgument(!this.startInstant.isEmpty() || !this.endInstant.isEmpty(),
"At least one of start and end instants should be specified.");
}
@@ -150,7 +150,7 @@ public abstract class InstantRange implements Serializable {
public ClosedClosedRangeNullableBoundary(String startInstant, String endInstant) {
super(startInstant, endInstant);
- ValidationUtils.checkArgument(!startInstant.isEmpty() || !endInstant.isEmpty(),
+ ValidationUtils.checkArgument(!this.startInstant.isEmpty() || !this.endInstant.isEmpty(),
"At least one of start and end instants should be specified.");
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java
index f99662780176..cfb9411be21f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v1/CompletionTimeQueryViewV1.java
@@ -30,14 +30,20 @@ import org.apache.hudi.common.util.VisibleForTesting;
import java.io.Serializable;
import java.time.Instant;
+import java.util.Collections;
+import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import static org.apache.hudi.common.table.read.IncrementalQueryAnalyzer.START_COMMIT_EARLIEST;
import static org.apache.hudi.common.table.timeline.InstantComparison.GREATER_THAN_OR_EQUALS;
import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN;
+import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN_OR_EQUALS;
public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Serializable {
private static final long serialVersionUID = 1L;
@@ -184,14 +190,20 @@ public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Seria
}
/**
- * Queries the instant start time with given completion time range.
+ * Queries the instant times within the given range.
*
- * @param timeline The timeline.
- * @param rangeStart The query range start completion time.
- * @param rangeEnd The query range end completion time.
+ * <p>A LAYOUT_VERSION_1 timeline (table versions 5 ~ 7) does not persist a separate completion
+ * time; an instant's completion time is, backward-compatibly, its instant (request) time. The
+ * range bounds are therefore matched against the requested time. Archived instants are loaded on
+ * demand when the requested range can overlap with the archived timeline. This is the "completion
+ * time = instant time" specialization of the same range logic implemented in
+ * {@code CompletionTimeQueryViewV2}.
+ *
+ * @param timeline The timeline (already filtered as per user configs by the caller).
+ * @param rangeStart The query range start requested time.
+ * @param rangeEnd The query range end requested time.
* @param rangeType The range type.
- * @param earliestInstantTimeFunc The function to generate the earliest start time boundary
- * with the minimum completion time.
+ * @param earliestInstantTimeFunc Unused for V1 and retained for API compatibility with V2.
*
* @return The sorted instant time list.
*/
@@ -201,7 +213,57 @@ public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Seria
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) {
+ return metaClient.getArchivedTimeline(startInstant.orElse(""), false)
+ .getCommitsTimeline()
+ .filterCompletedInstants()
+ .getInstantsAsStream()
+ .map(HoodieInstant::requestedTime);
}
// -------------------------------------------------------------------------
@@ -218,7 +280,7 @@ public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Seria
this.metaClient.getActiveTimeline()
.filterCompletedInstants().getInstantsAsStream()
.forEach(instant -> setCompletionTime(instant.requestedTime(), instant.getCompletionTime()));
- // Do not load archive timeline.
+ // The archived timeline is loaded on demand by range queries that can overlap with it.
}
private void setCompletionTime(String beginInstantTime, String completionTime) {
@@ -243,4 +305,4 @@ public class CompletionTimeQueryViewV1 implements CompletionTimeQueryView, Seria
public void close() {
this.beginToCompletionInstantTimeMap.clear();
}
-}
\ No newline at end of file
+}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/versioning/v1/TestCompletionTimeQueryViewV1.java b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/versioning/v1/TestCompletionTimeQueryViewV1.java
new file mode 100644
index 000000000000..c22a502b7546
--- /dev/null
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/versioning/v1/TestCompletionTimeQueryViewV1.java
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.common.table.timeline.versioning.v1;
+
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.InstantRange;
+import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
+import org.apache.hudi.common.table.timeline.HoodieArchivedTimeline;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.util.Option;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test cases for {@link CompletionTimeQueryViewV1#getInstantTimes}.
+ *
+ * <p>A LAYOUT_VERSION_1 timeline (table versions 5 ~ 7) does not persist a separate completion
+ * time; an instant's completion time is, backward-compatibly, its instant (request) time. The range
+ * query therefore filters by requested time. These tests assert the candidate instant selection
+ * matches the streaming/incremental read semantics (resume excludes the last issued offset, the
+ * first read includes the start commit, etc.), the same behavior as the V2 view keyed on completion
+ * time.
+ *
+ * <p>The active and archived timelines are mocked rather than materialized on disk because the
+ * {@code hudi-common} test scope has no hadoop storage implementation to build a real table from.
+ */
+public class TestCompletionTimeQueryViewV1 {
+
+ // Use datetime-format instant times so they are not short-circuited by the length < 10 branch in
+ // getCompletionTime, and behave like real V5 ~ V7 instants.
+ private static final String T1 = "20240101010001000";
+ private static final String T2 = "20240101010002000";
+ private static final String T3 = "20240101010003000";
+ private static final String T4 = "20240101010004000";
+ private static final String T5 = "20240101010005000";
+
+ private CompletionTimeQueryViewV1 view;
+ private HoodieTableMetaClient metaClient;
+ private HoodieTimeline timeline;
+
+ @BeforeEach
+ void setUp() {
+ List<HoodieInstant> archivedInstants = Arrays.asList(completedCommit(T1), completedCommit(T2));
+ List<HoodieInstant> activeInstants = Arrays.asList(completedCommit(T3), completedCommit(T4), completedCommit(T5));
+
+ timeline = completedTimelineMock(activeInstants);
+ HoodieTimeline archivedCommitsTimeline = completedTimelineMock(archivedInstants);
+ HoodieArchivedTimeline archivedTimeline = mock(HoodieArchivedTimeline.class);
+ when(archivedTimeline.getCommitsTimeline()).thenReturn(archivedCommitsTimeline);
+
+ // Minimal meta client wiring so the V1 view can be constructed (load()/cursor/firstNonSavepoint).
+ // Build the active timeline mock fully before stubbing getActiveTimeline(), otherwise Mockito
+ // sees nested stubbing (a when() started inside another when()'s argument) and fails.
+ HoodieActiveTimeline activeTimeline = activeTimelineMock(activeInstants);
+ metaClient = mock(HoodieTableMetaClient.class);
+ when(metaClient.getActiveTimeline()).thenReturn(activeTimeline);
+ when(metaClient.getArchivedTimeline(anyString(), eq(false))).thenReturn(archivedTimeline);
+ view = new CompletionTimeQueryViewV1(metaClient);
+ }
+
+ private static HoodieInstant completedCommit(String instantTime) {
+ // V1 instants are ordered/compared by requested(instant) time; completion time mirrors it.
+ return new HoodieInstant(
+ HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, instantTime, instantTime,
+ InstantComparatorV1.REQUESTED_TIME_BASED_COMPARATOR);
+ }
+
+ private static HoodieTimeline completedTimelineMock(List<HoodieInstant> instants) {
+ HoodieTimeline completedTimeline = mock(HoodieTimeline.class);
+ when(completedTimeline.filterCompletedInstants()).thenReturn(completedTimeline);
+ when(completedTimeline.getInstantsAsStream()).thenAnswer(invocation -> instants.stream());
+ when(completedTimeline.lastInstant()).thenAnswer(invocation -> Option.fromJavaOptional(
+ instants.stream().max(HoodieInstant::compareTo)));
+ return completedTimeline;
+ }
+
+ /**
+ * Builds a mock active timeline used only for constructing the view (load(), cursor and
+ * firstNonSavepointCommit). It is independent of the timeline passed to getInstantTimes.
+ */
+ private static HoodieActiveTimeline activeTimelineMock(List<HoodieInstant> instants) {
+ HoodieActiveTimeline active = mock(HoodieActiveTimeline.class);
+ when(active.filterCompletedInstants()).thenReturn(active);
+ when(active.getWriteTimeline()).thenReturn(active);
+ when(active.getInstantsAsStream()).thenAnswer(invocation -> instants.stream());
+ when(active.firstInstant()).thenAnswer(invocation -> Option.of(instants.get(0)));
+ when(active.getFirstNonSavepointCommit()).thenAnswer(invocation -> Option.of(instants.get(0)));
+ return active;
+ }
+
+ private List<String> query(Option<String> start, Option<String> end, InstantRange.RangeType rangeType) {
+ return view.getInstantTimes(timeline, start, end, rangeType);
+ }
+
+ @Test
+ void testClosedClosedRange() {
+ // [T2, T4]: first-time consume with an archived start and an active end, both bounds inclusive.
+ assertEquals(Arrays.asList(T2, T3, T4), query(Option.of(T2), Option.of(T4), InstantRange.RangeType.CLOSED_CLOSED));
+
+ // [T3, _): from the start commit to the latest, open end.
+ assertEquals(Arrays.asList(T3, T4, T5), query(Option.of(T3), Option.empty(), InstantRange.RangeType.CLOSED_CLOSED));
+
+ // [T1, T5]: the full range, also verifies the result is sorted ascending.
+ assertEquals(Arrays.asList(T1, T2, T3, T4, T5), query(Option.of(T1), Option.of(T5), InstantRange.RangeType.CLOSED_CLOSED));
+ }
+
+ @Test
+ void testOpenClosedResumeRange() {
+ // (T2, T4]: streaming resume from archived offset T2, the start point must be excluded.
+ assertEquals(Arrays.asList(T3, T4), query(Option.of(T2), Option.of(T4), InstantRange.RangeType.OPEN_CLOSED));
+
+ // (T4, _]: resume from T4 with open end, only T5 is new.
+ assertEquals(Collections.singletonList(T5), query(Option.of(T4), Option.empty(), InstantRange.RangeType.OPEN_CLOSED));
+
+ // (T5, _]: resume from the latest issued offset T5, nothing new, must not re-read T5.
+ assertEquals(Collections.emptyList(), query(Option.of(T5), Option.empty(), InstantRange.RangeType.OPEN_CLOSED));
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = InstantRange.RangeType.class, names = {"OPEN_CLOSED", "CLOSED_CLOSED"})
+ void testEarliestStart(InstantRange.RangeType rangeType) {
+ // ['earliest', T3]: archived instants must be included when consuming from the beginning.
+ assertEquals(Arrays.asList(T1, T2, T3),
+ query(Option.of("earliest"), Option.of(T3), rangeType));
+ }
+
+ @Test
+ void testEndOnlyReturnsLastAtOrBeforeEnd() {
+ // (_, T3]: no start commit, returns the single latest instant at or before T3.
+ assertEquals(Collections.singletonList(T3), query(Option.empty(), Option.of(T3), InstantRange.RangeType.CLOSED_CLOSED));
+
+ // (_, T5]: end at the latest, returns T5.
+ assertEquals(Collections.singletonList(T5), query(Option.empty(), Option.of(T5), InstantRange.RangeType.CLOSED_CLOSED));
+ }
+
+ @Test
+ void testEndOnlyFallsBackToArchivedCandidates() {
+ // The analyzer applies user-config filters to the archived timeline and then selects the last
+ // eligible instant, so the view must return all archived candidates at or before the end.
+ assertEquals(Arrays.asList(T1, T2),
+ query(Option.empty(), Option.of(T2), InstantRange.RangeType.CLOSED_CLOSED));
+ }
+
+ @Test
+ void testActiveRangeDoesNotLoadArchivedTimeline() {
+ assertEquals(Arrays.asList(T3, T4),
+ query(Option.of(T3), Option.of(T4), InstantRange.RangeType.CLOSED_CLOSED));
+ verify(metaClient, never()).getArchivedTimeline(anyString(), eq(false));
+ }
+
+ @Test
+ void testNoBoundsReadsLatestSnapshot() {
+ // (_, _): no range at all, reads the latest snapshot instant.
+ assertEquals(Collections.singletonList(T5), query(Option.empty(), Option.empty(), InstantRange.RangeType.CLOSED_CLOSED));
+ verify(metaClient, never()).getArchivedTimeline(anyString(), eq(false));
+ }
+}