This is an automated email from the ASF dual-hosted git repository.
yihua 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 9b39f0450787 fix(common): correct incremental fallback reads when a
stale non-write instant precedes the first active commit (#20273)
9b39f0450787 is described below
commit 9b39f0450787ac1265766eeed80be96f379e2c58
Author: Aditya Goenka <[email protected]>
AuthorDate: Wed Oct 14 10:09:54 2026 +0530
fix(common): correct incremental fallback reads when a stale non-write
instant precedes the first active commit (#20273)
Co-authored-by: Aditya Goenka <[email protected]>
---
.../timeline/TestCompletionTimeQueryView.java | 95 +++++++++++++++++
.../table/read/IncrementalQueryAnalyzer.java | 21 ++--
.../hudi/common/table/timeline/HoodieTimeline.java | 7 ++
.../versioning/v2/CompletionTimeQueryViewV2.java | 33 +++++-
.../hudi/core/read/BaseHoodieTableFileIndex.java | 21 +++-
.../core/read/BaseHoodieTableFileIndexTest.java | 73 ++++++++++++-
.../TestIncrementalReadWithFullTableScan.scala | 114 ++++++++++++++++++++-
7 files changed, 343 insertions(+), 21 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 93656debb7b3..c59fcaa3b8d1 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
@@ -19,12 +19,14 @@
package org.apache.hudi.client.timeline;
import org.apache.hudi.DummyActiveAction;
+import org.apache.hudi.avro.model.HoodieRollbackMetadata;
import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.model.HoodieCommitMetadata;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.log.InstantRange;
+import org.apache.hudi.common.table.read.IncrementalQueryAnalyzer;
import org.apache.hudi.common.table.timeline.ActiveAction;
import org.apache.hudi.common.table.timeline.CompletionTimeQueryView;
import org.apache.hudi.common.table.timeline.HoodieInstant;
@@ -202,6 +204,99 @@ public class TestCompletionTimeQueryView {
}
}
+ /**
+ * A completed non-write instant older than the archived commits (e.g. a
rollback the archiver has not
+ * reached, which survives an upgrade from table version 6) must not hide
those archived commits.
+ */
+ @Test
+ void testArchivedInstantsAfterStaleNonWriteInstant() throws Exception {
+ String tableName = "testTable";
+ String tablePath = tempFile.getAbsolutePath() + StoragePath.SEPARATOR +
tableName;
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(
+ HoodieTestUtils.getDefaultStorageConf(), tablePath,
HoodieTableType.COPY_ON_WRITE, tableName);
+ prepareTimeline(tablePath, metaClient);
+ // a completed rollback sorting between the archived commits 1 and 2
+ String staleRollback = "000000015";
+ HoodieTestTable testTable = HoodieTestTable.of(metaClient);
+ HoodieRollbackMetadata rollbackMetadata =
testTable.getRollbackMetadata("00000099", Collections.emptyMap(), false);
+ testTable.addRollback(staleRollback, rollbackMetadata,
null).addRollbackCompleted(staleRollback, rollbackMetadata, false);
+ metaClient.reloadActiveTimeline();
+
assertThat(metaClient.getActiveTimeline().firstInstant().get().requestedTime(),
is(staleRollback));
+
+ try (CompletionTimeQueryView view =
+
metaClient.getTableFormat().getTimelineFactory().createCompletionTimeQueryView(metaClient))
{
+ // 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"));
+ // a pending write instant is still pending
+ assertFalse(view.getCompletionTime(String.format("%08d",
11)).isPresent());
+ }
+ }
+
+ /**
+ * With archival beyond savepoint, a savepointed commit stays on the active
timeline while the commits after it are
+ * archived. That commit must not hide the archived commits after it.
+ */
+ @Test
+ void testArchivedInstantsAfterSavepointedCommit() throws Exception {
+ String tableName = "testTable";
+ String tablePath = tempFile.getAbsolutePath() + StoragePath.SEPARATOR +
tableName;
+ HoodieTableMetaClient metaClient = HoodieTestUtils.init(
+ HoodieTestUtils.getDefaultStorageConf(), tablePath,
HoodieTableType.COPY_ON_WRITE, tableName);
+ // archive 2 to 6, while 1 is savepointed and stays active
+ prepareTimeline(tablePath, metaClient, (writer, activeActions) ->
writer.write(activeActions.subList(1, 6), Option.empty(), Option.empty()));
+ String savepointed = String.format("%08d", 1);
+ HoodieTestTable testTable = HoodieTestTable.of(metaClient);
+ testTable.addCommit(savepointed, Option.of(String.format("%08d", 1001)),
+ Option.of(testTable.createCommitMetadata(savepointed,
WriteOperationType.INSERT, Arrays.asList("par1", "par2"), 10, false)));
+ testTable.addSavepoint(savepointed, Option.of(String.format("%08d",
1002)), testTable.getSavepointMetadata(savepointed, Collections.emptyMap()));
+ metaClient.reloadActiveTimeline();
+
assertThat(metaClient.getActiveTimeline().getWriteTimeline().firstInstant().get().requestedTime(),
is(savepointed));
+
+ try (CompletionTimeQueryView view =
+
metaClient.getTableFormat().getTimelineFactory().createCompletionTimeQueryView(metaClient))
{
+ // the savepointed commit is still found
+ assertThat(view.getCompletionTime(savepointed).orElse(""),
is(String.format("%08d", 1001)));
+ // archived commits after the savepointed commit 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"));
+ // and they are archived, while the savepointed commit and the commits
after the hole are active
+ assertTrue(view.isArchived(String.format("%08d", 4)));
+ assertFalse(view.isArchived(savepointed));
+ assertFalse(view.isArchived(String.format("%08d", 7)));
+ // a pending write instant is still pending
+ assertFalse(view.getCompletionTime(String.format("%08d",
11)).isPresent());
+ }
+
+ // an incremental query over the hole reads the archived commits from the
archived timeline
+ IncrementalQueryAnalyzer.QueryContext queryContext =
IncrementalQueryAnalyzer.builder()
+ .metaClient(metaClient)
+ .startCompletionTime(String.format("%08d", 1003))
+ .endCompletionTime(String.format("%08d", 1008))
+ .rangeType(InstantRange.RangeType.CLOSED_CLOSED)
+ .build()
+ .analyze();
+
assertThat(queryContext.getArchivedInstants().stream().map(HoodieInstant::requestedTime).collect(Collectors.joining(",")),
+ is("00000003,00000004,00000005,00000006"));
+
assertThat(queryContext.getActiveInstants().stream().map(HoodieInstant::requestedTime).collect(Collectors.joining(",")),
+ is("00000007,00000008"));
+
+ // a query that starts before the savepointed commit: the archived commits
after it are still read
+ queryContext = IncrementalQueryAnalyzer.builder()
+ .metaClient(metaClient)
+ .startCompletionTime(String.format("%08d", 1000))
+ .endCompletionTime(String.format("%08d", 1008))
+ .rangeType(InstantRange.RangeType.CLOSED_CLOSED)
+ .build()
+ .analyze();
+
assertThat(queryContext.getArchivedInstants().stream().map(HoodieInstant::requestedTime).collect(Collectors.joining(",")),
+ is("00000002,00000003,00000004,00000005,00000006"));
+
assertThat(queryContext.getActiveInstants().stream().map(HoodieInstant::requestedTime).collect(Collectors.joining(",")),
+ is("00000001,00000007,00000008"));
+ assertThat(String.join(",", queryContext.getInstantTimeList()),
+
is("00000001,00000002,00000003,00000004,00000005,00000006,00000007,00000008"));
+ }
+
@Test
void testGetInstantTimesWithOnlyEndCompletionTime() throws Exception {
String tableName = "testTable";
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java
index da0bfb567abe..bf2e78519fc3 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/read/IncrementalQueryAnalyzer.java
@@ -44,10 +44,10 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import java.util.stream.Stream;
/**
@@ -188,10 +188,12 @@ public class IncrementalQueryAnalyzer {
}
}
if (!archivedInstantTimes.isEmpty()) {
- archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.get(0));
+ archivedReadTimeline = getArchivedReadTimeline(metaClient,
archivedInstantTimes.stream().min(String::compareTo).get());
archivedInstants =
archivedReadTimeline.getInstantsAsStream().filter(instant ->
instantTimeSet.contains(instant.requestedTime())).collect(Collectors.toList());
}
- List<String> instants = Stream.concat(archivedInstants.stream(),
activeInstants.stream()).map(HoodieInstant::requestedTime).collect(Collectors.toList());
+ // archived and active instants can interleave (see
#splitInstantByActiveness), so order them by instant time.
+ List<String> instants = Stream.concat(archivedInstants.stream(),
activeInstants.stream()).map(HoodieInstant::requestedTime)
+ .sorted().collect(Collectors.toList());
if (instants.isEmpty()) {
// no instants completed within the give time range, returns early.
return QueryContext.EMPTY;
@@ -216,14 +218,11 @@ public class IncrementalQueryAnalyzer {
* Splits the given instant time list into a pair of archived instant list
and active instant list.
*/
private static Pair<List<String>, List<String>>
splitInstantByActiveness(List<String> instantTimeList, CompletionTimeQueryView
completionTimeQueryView) {
- int firstActiveIdx = IntStream.range(0, instantTimeList.size()).filter(i
->
!completionTimeQueryView.isArchived(instantTimeList.get(i))).findFirst().orElse(-1);
- if (firstActiveIdx == -1) {
- return Pair.of(instantTimeList, Collections.emptyList());
- } else if (firstActiveIdx == 0) {
- return Pair.of(Collections.emptyList(), instantTimeList);
- } else {
- return Pair.of(instantTimeList.subList(0, firstActiveIdx),
instantTimeList.subList(firstActiveIdx, instantTimeList.size()));
- }
+ // with archival beyond savepoint, a savepointed commit stays active while
the commits after it are archived,
+ // so the archived instants are not always a prefix of the list: check
each instant on its own.
+ Map<Boolean, List<String>> instantTimesByArchived =
instantTimeList.stream()
+
.collect(Collectors.partitioningBy(completionTimeQueryView::isArchived));
+ return Pair.of(instantTimesByArchived.get(true),
instantTimesByArchived.get(false));
}
private HoodieTimeline getFilteredTimeline(HoodieTableMetaClient metaClient)
{
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
index 4ad8ec769ac3..33d8b86be36b 100644
---
a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
+++
b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
@@ -82,6 +82,13 @@ public interface HoodieTimeline extends HoodieInstantReader,
Serializable {
COMMIT_ACTION, DELTA_COMMIT_ACTION, CLEAN_ACTION,
COMPACTION_ACTION, LOG_COMPACTION_ACTION, REPLACE_COMMIT_ACTION,
CLUSTERING_ACTION);
+ /**
+ * Write actions plus savepoint, used to find the first active write instant
that is not savepointed.
+ */
+ Set<String> WRITE_AND_SAVEPOINT_ACTIONS = CollectionUtils.createSet(
+ COMMIT_ACTION, DELTA_COMMIT_ACTION, COMPACTION_ACTION,
LOG_COMPACTION_ACTION,
+ REPLACE_COMMIT_ACTION, CLUSTERING_ACTION, SAVEPOINT_ACTION);
+
String COMMIT_EXTENSION = "." + COMMIT_ACTION;
String DELTA_COMMIT_EXTENSION = "." + DELTA_COMMIT_ACTION;
String CLEAN_EXTENSION = "." + CLEAN_ACTION;
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 be8dbd0462b1..f3126e9ee40c 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,6 +38,7 @@ 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;
@@ -83,6 +84,16 @@ public class CompletionTimeQueryViewV2 implements
CompletionTimeQueryView, Seria
*/
private final String firstNonSavepointCommit;
+ /**
+ * The initial cursor instant: write instants before it are archived, except
the savepointed ones.
+ */
+ private final String firstActiveWriteInstant;
+
+ /**
+ * The savepointed instants on the active timeline.
+ */
+ private final Set<String> activeSavepointedInstants;
+
/**
* The constructor.
*
@@ -91,9 +102,19 @@ public class CompletionTimeQueryViewV2 implements
CompletionTimeQueryView, Seria
public CompletionTimeQueryViewV2(HoodieTableMetaClient metaClient) {
this.metaClient = metaClient;
this.instantTimeToCompletionTimeMap = new ConcurrentHashMap<>();
- this.cursorInstant =
metaClient.getActiveTimeline().firstInstant().map(HoodieInstant::requestedTime).orElse("");
+ 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.
+ 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 =
metaClient.getActiveTimeline().getWriteTimeline().getFirstNonSavepointCommit().map(HoodieInstant::requestedTime).orElse("");
+ this.firstNonSavepointCommit =
activeTimeline.getWriteTimeline().getFirstNonSavepointCommit().map(HoodieInstant::requestedTime).orElse("");
load();
}
@@ -102,14 +123,18 @@ public class CompletionTimeQueryViewV2 implements
CompletionTimeQueryView, Seria
*/
public boolean isCompleted(String instantTime) {
// archival does not proceed beyond the first savepoint, so any instant
before that is completed.
- return this.instantTimeToCompletionTimeMap.containsKey(instantTime) ||
isArchived(instantTime);
+ return this.instantTimeToCompletionTimeMap.containsKey(instantTime)
+ || InstantComparison.compareTimestamps(instantTime, LESSER_THAN,
this.firstNonSavepointCommit);
}
/**
* Returns whether the instant is archived.
*/
public boolean isArchived(String instantTime) {
- return InstantComparison.compareTimestamps(instantTime, LESSER_THAN,
this.firstNonSavepointCommit);
+ // 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.
+ return InstantComparison.compareTimestamps(instantTime, LESSER_THAN,
this.firstActiveWriteInstant)
+ && !this.activeSavepointedInstants.contains(instantTime);
}
/**
diff --git
a/hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java
b/hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java
index 10a20083ddce..8e19f5491b39 100644
---
a/hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java
+++
b/hudi-common/src/main/java/org/apache/hudi/core/read/BaseHoodieTableFileIndex.java
@@ -44,6 +44,7 @@ import org.apache.hudi.common.util.DefaultSizeEstimator;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.common.util.collection.ExternalSpillableMap;
import org.apache.hudi.common.util.collection.ExternalSpillableMap.DiskMapType;
import org.apache.hudi.common.util.collection.Pair;
@@ -433,11 +434,23 @@ public abstract class BaseHoodieTableFileIndex implements
AutoCloseable {
}
private boolean isBeforeTimelineStarts() {
- if (isCompletionTimeBasedQuery) {
- return
metaClient.getActiveTimeline().isBeforeTimelineStartsByCompletionTime(incrementalQueryStartTime.get());
- } else {
- return
metaClient.getActiveTimeline().isBeforeTimelineStarts(incrementalQueryStartTime.get());
+ return isBeforeTimelineStarts(metaClient.getActiveTimeline(),
isCompletionTimeBasedQuery, incrementalQueryStartTime.get());
+ }
+
+ @VisibleForTesting
+ static boolean isBeforeTimelineStarts(HoodieTimeline activeTimeline, boolean
isCompletionTimeBasedQuery, String startTime) {
+ // Check against the same timeline findInstantsInRange() reads: the
incremental listing takes the
+ // written partitions from the write timeline, so it is only complete when
that timeline covers the
+ // start. Older non-write instants (e.g. a rollback the archiver has not
reached) must not count.
+ // Savepoints are kept so that savepointed commits left behind by archival
are skipped as the start.
+ HoodieTimeline writeTimeline =
activeTimeline.getTimelineOfActions(HoodieTimeline.WRITE_AND_SAVEPOINT_ACTIONS);
+ if
(writeTimeline.filterCompletedInstants().getFirstNonSavepointCommit().isEmpty())
{
+ // no completed write instant on the active timeline: the written
partitions are unknown, list them all
+ return true;
}
+ return isCompletionTimeBasedQuery
+ ? writeTimeline.isBeforeTimelineStartsByCompletionTime(startTime)
+ : writeTimeline.isBeforeTimelineStarts(startTime);
}
private HoodieTimeline findInstantsInRange() {
diff --git
a/hudi-common/src/test/java/org/apache/hudi/core/read/BaseHoodieTableFileIndexTest.java
b/hudi-common/src/test/java/org/apache/hudi/core/read/BaseHoodieTableFileIndexTest.java
index ee68443d7aab..a33a553e8bf2 100644
---
a/hudi-common/src/test/java/org/apache/hudi/core/read/BaseHoodieTableFileIndexTest.java
+++
b/hudi-common/src/test/java/org/apache/hudi/core/read/BaseHoodieTableFileIndexTest.java
@@ -19,11 +19,16 @@
package org.apache.hudi.core.read;
import org.apache.hudi.common.model.FileSlice;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.testutils.MockHoodieTimeline;
import org.apache.hudi.core.read.BaseHoodieTableFileIndex.PartitionPath;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
@@ -32,7 +37,9 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
+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.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -97,4 +104,68 @@ public class BaseHoodieTableFileIndexTest {
assertTrue(result.get(anotherEmpty).isEmpty(),
"Empty partition's file slice list must be present and empty (not
null, not missing)");
}
-}
\ No newline at end of file
+
+ /**
+ * The incremental partition listing reads written partitions from the write
timeline, so the
+ * "has the start been archived" guard must be evaluated against that same
timeline. A completed
+ * rollback (or clean) older than every active commit must not make an
archived start look active,
+ * otherwise only the partitions of the active commits in the range are
listed.
+ *
+ * <p>Timeline: rollback 001 (completed 002), commit 005 (completed 006),
commit 007 (completed 008).
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testIncrementalStartBeforeFirstWriteIsArchivedDespiteOlderNonWriteInstant(boolean
completionTimeBased) throws Exception {
+ HoodieTimeline activeTimeline = new MockHoodieTimeline(Arrays.asList(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.ROLLBACK_ACTION, "001", "002"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "005", "006"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "007", "008")));
+
+ // start after the rollback but before the first commit: archived
+ assertTrue(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, "003"));
+ // start at or after the first commit: covered by the active write timeline
+
assertFalse(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, completionTimeBased ? "006" : "005"));
+
assertFalse(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, "007"));
+ }
+
+ /**
+ * With archival beyond savepoint, a savepointed commit stays on the active
timeline while the commits after it are
+ * archived. A start in that hole is archived, so the savepointed commit
must not count as the start of the timeline.
+ *
+ * <p>Timeline: commit 001 (completed 002) with savepoint 001, commit 010
(completed 011), commit 012 (completed 013).
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testIncrementalStartInArchivalHoleAfterSavepointIsArchived(boolean
completionTimeBased) throws Exception {
+ HoodieTimeline activeTimeline = new MockHoodieTimeline(Arrays.asList(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "001", "002"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.SAVEPOINT_ACTION, "001", "003"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "010", "011"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.COMMIT_ACTION, "012", "013")));
+
+ // start between the savepointed commit and the first active commit after
it: archived
+ assertTrue(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, "005"));
+
assertFalse(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, completionTimeBased ? "011" : "010"));
+ }
+
+ /**
+ * Without a completed write instant on the active timeline the written
partitions are unknown, so every start is
+ * treated as archived and all partitions are listed.
+ */
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testIncrementalStartIsArchivedWithoutActiveWriteInstant(boolean
completionTimeBased) throws Exception {
+ HoodieTimeline activeTimeline = new MockHoodieTimeline(Arrays.asList(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.ROLLBACK_ACTION, "001", "002"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.CLEAN_ACTION, "003", "004")));
+
+ assertTrue(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, "000"));
+ assertTrue(BaseHoodieTableFileIndex.isBeforeTimelineStarts(activeTimeline,
completionTimeBased, "005"));
+
+ // a pending write instant alone does not tell which partitions were
written either
+ HoodieTimeline pendingOnlyTimeline = new MockHoodieTimeline(Arrays.asList(
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.COMPLETED,
HoodieTimeline.ROLLBACK_ACTION, "001", "002"),
+ INSTANT_GENERATOR.createNewInstant(HoodieInstant.State.REQUESTED,
HoodieTimeline.COMPACTION_ACTION, "003")));
+
assertTrue(BaseHoodieTableFileIndex.isBeforeTimelineStarts(pendingOnlyTimeline,
completionTimeBased, "005"));
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
index 6c26e4fe1062..16556473e507 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/functional/TestIncrementalReadWithFullTableScan.scala
@@ -21,7 +21,7 @@ import org.apache.hudi.{DataSourceReadOptions,
DataSourceWriteOptions, HoodieSpa
import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.common.model.HoodieTableType
import org.apache.hudi.common.table.HoodieTableConfig
-import org.apache.hudi.common.table.timeline.{HoodieInstant,
HoodieInstantTimeGenerator, InstantComparison}
+import org.apache.hudi.common.table.timeline.{HoodieInstant,
HoodieInstantTimeGenerator, HoodieTimeline, InstantComparison}
import
org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator.instantTimeMinusMillis
import
org.apache.hudi.common.table.timeline.InstantComparison.compareTimestamps
import
org.apache.hudi.common.testutils.HoodieTestDataGenerator.recordsToStrings
@@ -30,6 +30,7 @@ import org.apache.hudi.testutils.HoodieSparkClientTestBase
import org.apache.spark.SparkException
import org.apache.spark.sql.{SaveMode, SparkSession}
+import org.apache.spark.sql.functions.{assert_true, col, lit, min, when}
import org.junit.jupiter.api.{AfterEach, BeforeEach}
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows,
assertTrue}
import org.junit.jupiter.api.function.Executable
@@ -261,6 +262,117 @@ class TestIncrementalReadWithFullTableScan extends
HoodieSparkClientTestBase {
"a span starting at 000 returns every record on both table versions")
}
+ /**
+ * A completed non-commit instant (here a rollback of a failed write) that
sits on the active timeline
+ * ahead of the first active commit must not change the result of a fallback
full table scan.
+ *
+ * The table version 6 archiver only archives rollback instants once there
are more of them than
+ * {@code hoodie.keep.max.commits}, so a single old rollback outlives the
commits around it, and it
+ * stays on the active timeline when the table is upgraded to version 9. The
relation then sees the
+ * start instant as archived (it checks the commit timeline) and falls back
to a full table scan,
+ * while the file index used to check the whole active timeline, saw the
start as not archived and
+ * listed only the partitions written by active commits in the range.
+ *
+ * Commit k writes 100 records to partition p(k % 5), tagged with commit_no
= k:
+ * {{{
+ * C1 RB C2 ... C6 ... | ... C12 [C13 at version 9]
+ * ^archived ^stale | active
+ * }}}
+ */
+ @ParameterizedTest
+ @CsvSource(value = Array(
+ "COPY_ON_WRITE,false",
+ "COPY_ON_WRITE,true",
+ "MERGE_ON_READ,false"))
+ def testFallbackFullTableScanWithStaleRollbackOnActiveTimeline(tableType:
String, upgradeToV9: Boolean): Unit = {
+ def opts(tableVersion: Int) = Map(
+ "hoodie.insert.shuffle.parallelism" -> "2",
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "id",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "ts",
+ HoodieWriteConfig.TBL_NAME.key -> "hoodie_test",
+ HoodieMetadataConfig.COMPACT_NUM_DELTA_COMMITS.key -> "1",
+ DataSourceWriteOptions.TABLE_TYPE.key -> tableType,
+ HoodieWriteConfig.WRITE_TABLE_VERSION.key -> tableVersion.toString,
+ HoodieWriteConfig.AUTO_UPGRADE_VERSION.key -> (tableVersion >
6).toString,
+ DataSourceWriteOptions.OPERATION.key ->
DataSourceWriteOptions.INSERT_OPERATION_OPT_VAL,
+ "hoodie.clean.commits.retained" -> "1",
+ "hoodie.keep.min.commits" -> "2",
+ "hoodie.keep.max.commits" -> "3")
+
+ def batch(k: Int) = spark.range(k * 1000L, k * 1000L + perBatchSize)
+ .withColumn("partition", lit(s"p${k % 5}"))
+ .withColumn("commit_no", lit(k))
+ .withColumn("ts", lit(k.toLong))
+
+ def write(k: Int, tableVersion: Int): Unit =
+
batch(k).write.format("org.apache.hudi").options(opts(tableVersion)).mode(SaveMode.Append).save(basePath)
+
+ for (k <- 1 to 12) {
+ if (k == 2) {
+ // fails inside the write stages, leaving a pending instant that the
next write rolls back
+ assertThrows(classOf[Exception], new Executable {
+ override def execute(): Unit = batch(99)
+ .withColumn("commit_no", when(assert_true(col("id") < 0).isNull,
col("commit_no")))
+
.write.format("org.apache.hudi").options(opts(6)).mode(SaveMode.Append).save(basePath)
+ })
+ }
+ write(k, 6)
+ }
+ val numCommits = if (upgradeToV9) {
+ write(13, 9)
+ 13
+ } else {
+ 12
+ }
+
+ val metaClient = createMetaClient(spark, basePath)
+ val tableVersion = metaClient.getTableConfig.getTableVersion.versionCode()
+ assertEquals(if (upgradeToV9) 9 else 6, tableVersion)
+ val firstInstant =
metaClient.getActiveTimeline.filterCompletedInstants().firstInstant().get()
+ val firstCommit =
metaClient.getCommitsTimeline.filterCompletedInstants().firstInstant().get()
+ assertEquals(HoodieTimeline.ROLLBACK_ACTION, firstInstant.getAction,
+ "the rollback of the failed write must still be the first instant on the
active timeline")
+
+ // instants of the data commits, keyed by commit_no (requested time =
_hoodie_commit_time)
+ val requested: Map[Int, String] =
spark.read.format("org.apache.hudi").load(basePath)
+ .groupBy("commit_no").agg(min("_hoodie_commit_time")).collect()
+ .map(r => r.getInt(0) -> r.getString(1)).toMap
+ lazy val completion: Map[Int, String] = {
+ val byRequested = (metaClient.getArchivedTimeline.getInstants.asScala ++
+
metaClient.getActiveTimeline.filterCompletedInstants().getInstants.asScala)
+ .map(i => i.requestedTime -> i.getCompletionTime).toMap
+ requested.map { case (k, t) => k -> byRequested(t) }
+ }
+ assertTrue(compareTimestamps(requested(6), InstantComparison.LESSER_THAN,
firstCommit.requestedTime),
+ "commits up to C6 must be archived")
+ assertTrue(compareTimestamps(requested(11),
InstantComparison.GREATER_THAN_OR_EQUALS, firstCommit.requestedTime),
+ "C11 must still be active")
+
+ // V1 ranges over requested time, V2 over completion time
+ def ts(k: Int): String = if (tableVersion < 8) requested(k) else
completion(k)
+
+ def incremental(startK: Int, endK: Option[Int], fallback: Boolean):
Seq[(Int, String)] = {
+ var reader = spark.read.format("org.apache.hudi")
+ .option(DataSourceReadOptions.QUERY_TYPE.key,
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL)
+ .option(DataSourceReadOptions.START_COMMIT.key, ts(startK))
+
.option(DataSourceReadOptions.INCREMENTAL_FALLBACK_TO_FULL_TABLE_SCAN.key,
fallback.toString)
+ endK.foreach(e => reader =
reader.option(DataSourceReadOptions.END_COMMIT.key, ts(e)))
+ reader.load(basePath).select("commit_no", "partition").collect().map(r
=> (r.getInt(0), r.getString(1))).toSeq
+ }
+
+ def expected(startK: Int, endK: Option[Int]): Seq[(Int, String)] =
+ ((startK + 1) to endK.getOrElse(numCommits)).flatMap(k =>
Seq.fill(perBatchSize)((k, s"p${k % 5}")))
+
+ for ((startK, endK) <- Seq((3, Some(6)), (3, None), (11, None))) {
+ val window = s"C$startK -> ${endK.map("C" + _).getOrElse("latest")}"
+ assertEquals(expected(startK, endK).sorted, incremental(startK, endK,
fallback = true).sorted,
+ s"fallback full table scan over $window")
+ }
+ // fully active window: no fallback needed
+ assertEquals(expected(11, None).sorted, incremental(11, None, fallback =
false).sorted)
+ }
+
private def runIncrementalQueryAndCompare(
startTs: String,
endTs: String,
