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 <[email protected]> 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 <[email protected]> (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)); + } +}
