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/15 09:38:55 UTC
(hudi) branch master updated: fix(common): avoid archive lookups for active clean and rollback instants (#20340)
This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 75fb5d5ecd92 fix(common): avoid archive lookups for active clean and rollback instants (#20340)
75fb5d5ecd92 is described below
commit 75fb5d5ecd922ed873bcd6bb31382b6630f36e67
Author: Danny Chan <yu...@gmail.com>
AuthorDate: Thu Oct 15 17:38:48 2026 +0800
fix(common): avoid archive lookups for active clean and rollback instants (#20340)
---
.../timeline/TestCompletionTimeQueryView.java | 3 ++
.../versioning/v2/CompletionTimeQueryViewV2.java | 21 ++------
.../table/read/TestIncrementalQueryAnalyzer.java | 59 ++++++++++++++++++++++
3 files changed, 67 insertions(+), 16 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/timeline/TestCompletionTimeQueryView.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/timeline/TestCompletionTimeQueryView.java
index c59fcaa3b8d1..ac568c98cbb8 100644
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/timeline/TestCompletionTimeQueryView.java
+++ b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/timeline/TestCompletionTimeQueryView.java
@@ -225,6 +225,9 @@ public class TestCompletionTimeQueryView {
try (CompletionTimeQueryView view =
metaClient.getTableFormat().getTimelineFactory().createCompletionTimeQueryView(metaClient)) {
+ assertFalse(view.isArchived(staleRollback));
+ assertTrue(view.isArchived(String.format("%08d", 4)));
+ assertFalse(view.isArchived(String.format("%08d", 7)));
// archived commits newer than the rollback are found, not reported as pending
assertThat(view.getCompletionTime(String.format("%08d", 4)).orElse(""), is(String.format("%08d", 1004)));
assertThat(getInstantTimeSetFormattedString(view, 3 + 1000, 6 + 1000), is("00000003,00000004,00000005,00000006"));
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java
index f3126e9ee40c..fc7fa7340d8b 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/versioning/v2/CompletionTimeQueryViewV2.java
@@ -38,7 +38,6 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Function;
@@ -89,11 +88,6 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria
*/
private final String firstActiveWriteInstant;
- /**
- * The savepointed instants on the active timeline.
- */
- private final Set<String> activeSavepointedInstants;
-
/**
* The constructor.
*
@@ -103,15 +97,11 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria
this.metaClient = metaClient;
this.instantTimeToCompletionTimeMap = new ConcurrentHashMap<>();
HoodieTimeline activeTimeline = metaClient.getActiveTimeline();
- // Instants at or after the cursor are assumed to be on the active timeline, so it must start at the first
- // write instant that is not savepointed: an older non-write instant (e.g. a rollback the archiver has not reached)
- // or a savepointed commit left behind by archival would otherwise hide the archived write instants after it.
- // Instants before the cursor that are still active are loaded eagerly by #load. Without such a write instant,
- // nothing after the last active instant can be archived.
+ // Tables upgraded from pre-v8 can have interleaved archived and active instants,
+ // start at the first non-savepointed write so older active instants do not hide archived writes,
+ // fall back to the last active instant when no such write exists.
this.firstActiveWriteInstant = activeTimeline.getTimelineOfActions(HoodieTimeline.WRITE_AND_SAVEPOINT_ACTIONS).getFirstNonSavepointCommit()
.or(activeTimeline::lastInstant).map(HoodieInstant::requestedTime).orElse("");
- this.activeSavepointedInstants = activeTimeline.getSavePointTimeline().getInstantsAsStream()
- .map(HoodieInstant::requestedTime).collect(Collectors.toSet());
this.cursorInstant = this.firstActiveWriteInstant;
// Note: use getWriteTimeline() to keep sync with the fs view visibleCommitsAndCompactionTimeline, see AbstractTableFileSystemView.refreshTimeline.
this.firstNonSavepointCommit = activeTimeline.getWriteTimeline().getFirstNonSavepointCommit().map(HoodieInstant::requestedTime).orElse("");
@@ -131,10 +121,9 @@ public class CompletionTimeQueryViewV2 implements CompletionTimeQueryView, Seria
* Returns whether the instant is archived.
*/
public boolean isArchived(String instantTime) {
- // with archival beyond savepoint, the write instants between a savepointed commit and the first active write
- // instant are archived as well, while the savepointed commit itself stays active.
+ // A savepointed write or an older clean/rollback can still be active before the write boundary.
return InstantComparison.compareTimestamps(instantTime, LESSER_THAN, this.firstActiveWriteInstant)
- && !this.activeSavepointedInstants.contains(instantTime);
+ && !this.metaClient.getActiveTimeline().containsInstant(instantTime);
}
/**
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestIncrementalQueryAnalyzer.java b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestIncrementalQueryAnalyzer.java
index 4ed4c34e41e0..061fa194147a 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestIncrementalQueryAnalyzer.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestIncrementalQueryAnalyzer.java
@@ -21,21 +21,35 @@ package org.apache.hudi.common.table.read;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.InstantRange;
+import org.apache.hudi.common.table.timeline.ArchivedTimelineLoader;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.table.timeline.versioning.v2.CompletionTimeQueryViewV2;
+import org.apache.hudi.common.testutils.MockHoodieTimeline;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import java.util.stream.Stream;
+import static org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
class TestIncrementalQueryAnalyzer {
@@ -77,6 +91,51 @@ class TestIncrementalQueryAnalyzer {
assertThrows(IllegalStateException.class, IncrementalQueryAnalyzer.QueryContext.EMPTY::getLastInstant);
}
+ @ParameterizedTest
+ @CsvSource({"clean,false", "clean,true", "rollback,false", "rollback,true"})
+ void testActiveNonWriteDoesNotTriggerArchivedTimelineLookup(String action, boolean savepoint) {
+ String savepointTime = "20261001000000000";
+ String nonWriteTime = "20261002000000000";
+ String writeTime = "20261003000000000";
+ String completionTime = "20261005000000000";
+ List<HoodieInstant> instants = new ArrayList<>();
+ if (savepoint) {
+ instants.add(INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, savepointTime, savepointTime));
+ instants.add(INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.SAVEPOINT_ACTION, savepointTime, savepointTime));
+ }
+ // Both actions finish in the query window, but only the write belongs in the result.
+ instants.add(INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED, action, nonWriteTime, completionTime));
+ instants.add(INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED, HoodieTimeline.COMMIT_ACTION, writeTime, completionTime));
+ ArchivedTimelineLoader loader = mock(ArchivedTimelineLoader.class);
+ MockHoodieTimeline active = new MockHoodieTimeline(instants);
+ HoodieTableMetaClient metaClient = mock(HoodieTableMetaClient.class, RETURNS_DEEP_STUBS);
+ when(metaClient.getActiveTimeline()).thenReturn(active);
+ when(metaClient.getCommitsTimeline()).thenReturn(active.getCommitsTimeline());
+ when(metaClient.getCommitsAndCompactionTimeline()).thenReturn(active.getCommitsAndCompactionTimeline());
+ when(metaClient.getTableFormat().getTimelineFactory().createArchivedTimelineLoader()).thenReturn(loader);
+ try (CompletionTimeQueryViewV2 view = new CompletionTimeQueryViewV2(metaClient)) {
+ assertEquals(writeTime, view.getCursorInstant());
+ assertFalse(view.isArchived(nonWriteTime));
+ assertFalse(view.isArchived(writeTime));
+ if (savepoint) {
+ assertFalse(view.isArchived(savepointTime));
+ }
+ when(metaClient.getTableFormat().getTimelineFactory().createCompletionTimeQueryView(metaClient)).thenReturn(view);
+ IncrementalQueryAnalyzer.QueryContext result = IncrementalQueryAnalyzer.builder()
+ .metaClient(metaClient)
+ .startCompletionTime(completionTime)
+ .endCompletionTime(completionTime)
+ .rangeType(InstantRange.RangeType.CLOSED_CLOSED)
+ .build().analyze();
+ assertEquals(Collections.singletonList(writeTime), result.getInstantTimeList());
+ assertTrue(result.getArchivedInstants().isEmpty());
+ // The one-day lookback is after writeTime, isolating classification from archive completion-time loading.
+ verifyNoInteractions(loader);
+ verify(metaClient, never()).getArchivedTimeline(anyString(), eq(false));
+ verify(metaClient, never()).getArchivePath();
+ }
+ }
+
@Test
void testBuilderRequiresMetaClientAndRangeType() {
assertThrows(NullPointerException.class, () -> IncrementalQueryAnalyzer.builder()