Skip to content

[HUDI-6678] Fix the acquisition of clean&rollback instants to archive #9416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yihua merged 3 commits into from
Sep 16, 2023
Merged

[HUDI-6678] Fix the acquisition of clean&rollback instants to archive #9416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ff356a7
[HUDI-6678] Fix the acquisition of clean and rollback instants to arc…
Zouxxyy Sep 14, 2023
d167baf
for test case
Zouxxyy Sep 14, 2023
1d98932
rename oldest to earliest
Zouxxyy Sep 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
244 changes: 126 additions & 118 deletions ...i-client-common/src/main/java/org/apache/hudi/client/timeline/HoodieTimelineArchiver.java
View file
Open in desktop

Large diffs are not rendered by default.

2 changes: 1 addition & 1 deletion ...ent/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -506,7 +506,7 @@ public Pair<Boolean, List<CleanFileInfo>> getDeletePaths(String partitionPath, O
}

/**
* Returns earliest commit to retain based on cleaning policy.
* Returns the earliest commit to retain based on cleaning policy.
*/
public Option<HoodieInstant> getEarliestCommitToRetain() {
return CleanerUtils.getEarliestCommitToRetain(
Expand Down
176 changes: 69 additions & 107 deletions ...client/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiver.java
View file
Open in desktop

Large diffs are not rendered by default.

8 changes: 4 additions & 4 deletions hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,14 +236,14 @@ public static boolean isPendingClusteringInstant(HoodieTableMetaClient metaClien
}

/**
* Returns the oldest instant to retain.
* Make sure the clustering instant won't be archived before cleaned, and the oldest inflight clustering instant has a previous commit.
* Returns the earliest instant to retain.
* Make sure the clustering instant won't be archived before cleaned, and the earliest inflight clustering instant has a previous commit.
*
* @param activeTimeline The active timeline
* @param metaClient The meta client
* @return the oldest instant to retain for clustering
* @return the earliest instant to retain for clustering
*/
public static Option<HoodieInstant> getOldestInstantToRetainForClustering(
public static Option<HoodieInstant> getEarliestInstantToRetainForClustering(
HoodieActiveTimeline activeTimeline, HoodieTableMetaClient metaClient) throws IOException {
Option<HoodieInstant> oldestInstantToRetain = Option.empty();
HoodieTimeline replaceTimeline = activeTimeline.getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.REPLACE_COMMIT_ACTION));
Expand Down
6 changes: 3 additions & 3 deletions hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -339,7 +339,7 @@ public static Option<Pair<HoodieTimeline, HoodieInstant>> getDeltaCommitsSinceLa
}

/**
* Gets the oldest instant to retain for MOR compaction.
* Gets the earliest instant to retain for MOR compaction.
* If there is no completed compaction,
* num delta commits >= "hoodie.compact.inline.max.delta.commits"
* If there is a completed compaction,
Expand All @@ -348,9 +348,9 @@ public static Option<Pair<HoodieTimeline, HoodieInstant>> getDeltaCommitsSinceLa
* @param activeTimeline Active timeline of a table.
* @param maxDeltaCommits Maximum number of delta commits that trigger the compaction plan,
* i.e., "hoodie.compact.inline.max.delta.commits".
* @return the oldest instant to keep for MOR compaction.
* @return the earliest instant to keep for MOR compaction.
*/
public static Option<HoodieInstant> getOldestInstantToRetainForCompaction(
public static Option<HoodieInstant> getEarliestInstantToRetainForCompaction(
HoodieActiveTimeline activeTimeline, int maxDeltaCommits) {
Option<Pair<HoodieTimeline, HoodieInstant>> deltaCommitsInfoOption =
CompactionUtils.getDeltaCommitsSinceLatestCompaction(activeTimeline);
Expand Down
6 changes: 3 additions & 3 deletions hudi-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ public void testGetOldestInstantToRetainForClustering() throws IOException {
HoodieInstant inflightInstant3 = metaClient.getActiveTimeline().transitionReplaceRequestedToInflight(requestedInstant3, Option.empty());
HoodieInstant completedInstant3 = metaClient.getActiveTimeline().transitionReplaceInflightToComplete(inflightInstant3, Option.empty());
metaClient.reloadActiveTimeline();
Option<HoodieInstant> actual = ClusteringUtils.getOldestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
Option<HoodieInstant> actual = ClusteringUtils.getEarliestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
assertTrue(actual.isPresent());
assertEquals(clusterTime1, actual.get().getTimestamp(), "no clean in timeline, retain first replace commit");

Expand All @@ -168,7 +168,7 @@ public void testGetOldestInstantToRetainForClustering() throws IOException {
metaClient.getActiveTimeline().transitionCleanInflightToComplete(inflightInstant4,
TimelineMetadataUtils.serializeCleanMetadata(cleanMetadata));
metaClient.reloadActiveTimeline();
actual = ClusteringUtils.getOldestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
actual = ClusteringUtils.getEarliestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
assertEquals(clusterTime3, actual.get().getTimestamp(),
"retain the first replace commit after the earliestInstantToRetain ");
}
Expand Down Expand Up @@ -206,7 +206,7 @@ public void testGetOldestInstantToRetainForClusteringKeepFileVersion() throws IO
metaClient.getActiveTimeline().transitionReplaceInflightToComplete(inflightInstant3, Option.empty());
metaClient.reloadActiveTimeline();

Option<HoodieInstant> actual = ClusteringUtils.getOldestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
Option<HoodieInstant> actual = ClusteringUtils.getEarliestInstantToRetainForClustering(metaClient.getActiveTimeline(), metaClient);
assertEquals(clusterTime2, actual.get().getTimestamp(),
"retain the first replace commit after the last complete clean ");
}
Expand Down
8 changes: 4 additions & 4 deletions hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,25 +285,25 @@ public void testGetDeltaCommitsSinceLatestCompactionWithEmptyDeltaCommits() {
@ValueSource(booleans = {true, false})
public void testGetOldestInstantToKeepForCompaction(boolean hasCompletedCompaction) {
HoodieActiveTimeline timeline = prepareTimeline(hasCompletedCompaction);
Option<HoodieInstant> actual = CompactionUtils.getOldestInstantToRetainForCompaction(timeline, 20);
Option<HoodieInstant> actual = CompactionUtils.getEarliestInstantToRetainForCompaction(timeline, 20);

if (hasCompletedCompaction) {
assertEquals(new HoodieInstant(false, HoodieTimeline.COMMIT_ACTION, "06"), actual.get());
} else {
assertEquals(new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "01"), actual.get());
}

actual = CompactionUtils.getOldestInstantToRetainForCompaction(timeline, 3);
actual = CompactionUtils.getEarliestInstantToRetainForCompaction(timeline, 3);
assertEquals(new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "07"), actual.get());

actual = CompactionUtils.getOldestInstantToRetainForCompaction(timeline, 2);
actual = CompactionUtils.getEarliestInstantToRetainForCompaction(timeline, 2);
assertEquals(new HoodieInstant(false, HoodieTimeline.DELTA_COMMIT_ACTION, "08"), actual.get());
}

@Test
public void testGetOldestInstantToKeepForCompactionWithEmptyDeltaCommits() {
HoodieActiveTimeline timeline = new MockHoodieActiveTimeline();
assertEquals(Option.empty(), CompactionUtils.getOldestInstantToRetainForCompaction(timeline, 20));
assertEquals(Option.empty(), CompactionUtils.getEarliestInstantToRetainForCompaction(timeline, 20));
}

private HoodieActiveTimeline prepareTimeline(boolean hasCompletedCompaction) {
Expand Down