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