Skip to content

[HUDI-2892][BUG]Pending Clustering may stain the ActiveTimeLine and lead to incomplete query results #4172

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 7 commits into from
Dec 10, 2021
Merged

[HUDI-2892][BUG]Pending Clustering may stain the ActiveTimeLine and lead to incomplete query results #4172

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
365c582
pending clustering will block archive
Nov 30, 2021
bb29721
merge from master
Dec 1, 2021
8740fec
code review
Dec 6, 2021
f65e697
code review
Dec 6, 2021
03e9cac
code review
Dec 9, 2021
81698b8
code review
Dec 9, 2021
76edd1f
merge from master
Dec 9, 2021
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
6 changes: 3 additions & 3 deletions ...ient/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTimelineArchiveLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,8 +158,8 @@ private Stream<HoodieInstant> getCommitInstantsToArchive() {
// TODO (na) : Add a way to return actions associated with a timeline and then merge/unify
// with logic above to avoid Stream.concats
HoodieTimeline commitTimeline = table.getCompletedCommitsTimeline();
Option<HoodieInstant> oldestPendingCompactionInstant =
table.getActiveTimeline().filterPendingCompactionTimeline().firstInstant();
Option<HoodieInstant> oldestPendingCompactionAndReplaceInstant =
table.getActiveTimeline().filterPendingCompactionAndReplaceTimeline().firstInstant();

@yihua yihua Dec 6, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

possible to use filterInflightsAndRequested() instead of creating a new API filterPendingCompactionAndReplaceTimeline()?

@zhangyue19921010 zhangyue19921010 Dec 9, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure thing, changed. Maybe use !s.isComplete same as before is more accurate than filterInflightsAndRequested() :>

Option<HoodieInstant> oldestInflightCommitInstant =
table.getActiveTimeline()
.getTimelineOfActions(CollectionUtils.createSet(HoodieTimeline.COMMIT_ACTION, HoodieTimeline.DELTA_COMMIT_ACTION))
Expand All @@ -176,7 +176,7 @@ private Stream<HoodieInstant> getCommitInstantsToArchive() {
return !(firstSavepoint.isPresent() && HoodieTimeline.compareTimestamps(firstSavepoint.get().getTimestamp(), LESSER_THAN_OR_EQUALS, s.getTimestamp()));
}).filter(s -> {
// Ensure commits >= oldest pending compaction commit is retained
return oldestPendingCompactionInstant
return oldestPendingCompactionAndReplaceInstant
.map(instant -> HoodieTimeline.compareTimestamps(instant.getTimestamp(), GREATER_THAN, s.getTimestamp()))
.orElse(true);
});
Expand Down
19 changes: 19 additions & 0 deletions ...ient/hudi-spark-client/src/test/java/org/apache/hudi/io/TestHoodieTimelineArchiveLog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,25 @@ public void testArchiveCommitSavepointNoHole() throws Exception {
"Archived commits should always be safe");
}

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testPendingClusteringWillBlockArchival(boolean enableMetadata) throws Exception {
HoodieWriteConfig writeConfig = initTestTableAndGetWriteConfig(enableMetadata, 2, 5, 2);
HoodieTestDataGenerator.createPendingReplaceFile(basePath, "00000000", wrapperFs.getConf());
for (int i = 1; i < 8; i++) {
testTable.doWriteOperation("0000000" + i, WriteOperationType.UPSERT, Arrays.asList("p1", "p2"), Arrays.asList("p1", "p2"), 2);
// archival
Pair<List<HoodieInstant>, List<HoodieInstant>> commitsList = archiveAndGetCommitsList(writeConfig);
List<HoodieInstant> originalCommits = commitsList.getKey();
List<HoodieInstant> commitsAfterArchival = commitsList.getValue();
assertEquals(originalCommits, commitsAfterArchival);
}

HoodieTimeline timeline = metaClient.getActiveTimeline().reload().getCommitsTimeline().filterCompletedInstants();
assertEquals(7, timeline.countInstants(),
"Since we have a pending clustering instant at 00000000, we should never archive any commit after 00000000");
}

@ParameterizedTest
@ValueSource(booleans = {true, false})
public void testArchiveRollbacksTestTable(boolean enableMetadata) throws Exception {
Expand Down
8 changes: 8 additions & 0 deletions hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieDefaultTimeline.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,14 @@ public HoodieTimeline filterPendingCompactionTimeline() {
instants.stream().filter(s -> s.getAction().equals(HoodieTimeline.COMPACTION_ACTION) && !s.isCompleted()), details);
}

@Override
public HoodieTimeline filterPendingCompactionAndReplaceTimeline() {

@yihua yihua Dec 6, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not required in this PR. In general, should we have something like filterPendingTableServicesTimeline() so that any new async table service should not cause the same problem of staining the timeline again? or maybe filter !s.isCompleted() only?

As a side note, the COMPACTION_ACTION only appears if it is not complete. Once the compaction is complete, it shows as a completed commit instead of completed compaction in the timeline.

@zhangyue19921010 zhangyue19921010 Dec 9, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice idea.
For now there are three kinds of table services including CLUSTER, COMPACT and CLEAN . And only commit service could stain the active timeline.
With the continuous development of HUDI, there may be more and more types of commits service. It's better to put some restrictions in case that stain the active timeline by accident.
I may spend time on this. Maybe contribute a another PR to make it happen :)

return new HoodieDefaultTimeline(
instants.stream()
.filter(s -> s.getAction().equals(HoodieTimeline.COMPACTION_ACTION) || (s.getAction().equals(HoodieTimeline.REPLACE_COMMIT_ACTION) && !s.isCompleted())),
details);
}

@Override
public HoodieDefaultTimeline findInstantsInRange(String startTs, String endTs) {
return new HoodieDefaultTimeline(
Expand Down
7 changes: 7 additions & 0 deletions hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieTimeline.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,13 @@ public interface HoodieTimeline extends Serializable {
*/
HoodieTimeline filterPendingCompactionTimeline();

/**
* Filter this timeline to just include requested and inflight compaction/replace instants.
*
* @return
*/
HoodieTimeline filterPendingCompactionAndReplaceTimeline();

/**
* Filter this timeline to just include requested and inflight replacecommit instants.
*/
Expand Down
11 changes: 11 additions & 0 deletions hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -388,6 +388,17 @@ public static void createReplaceFile(String basePath, String instantTime, Config
.forEach(f -> createMetadataFile(f, basePath, configuration, commitMetadata));
}

public static void createPendingReplaceFile(String basePath, String instantTime, Configuration configuration, HoodieCommitMetadata commitMetadata) {
Arrays.asList(HoodieTimeline.makeInflightReplaceFileName(instantTime),
HoodieTimeline.makeRequestedReplaceFileName(instantTime))
.forEach(f -> createMetadataFile(f, basePath, configuration, commitMetadata));
}

public static void createPendingReplaceFile(String basePath, String instantTime, Configuration configuration) {
HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata();
createPendingReplaceFile(basePath, instantTime, configuration, commitMetadata);
}

public static void createEmptyCleanRequestedFile(String basePath, String instantTime, Configuration configuration)
throws IOException {
Path commitFile = new Path(basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
Expand Down