yihua commented on code in PR #9039:
URL: https://github.com/apache/hudi/pull/9039#discussion_r1240218048
##########
hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java:
##########
@@ -345,22 +345,34 @@ public static Option<Pair<HoodieTimeline, HoodieInstant>>
getDeltaCommitsSinceLa
* @return the oldest instant to keep for MOR compaction.
*/
public static Option<HoodieInstant> getOldestInstantToRetainForCompaction(
- HoodieActiveTimeline activeTimeline, int maxDeltaCommits) {
+ HoodieActiveTimeline activeTimeline, HoodieTableMetaClient metaClient,
int maxDeltaCommits) throws IOException {
+ Option<HoodieInstant> oldestDeltaCommitToRetain = Option.empty();
Option<Pair<HoodieTimeline, HoodieInstant>> deltaCommitsInfoOption =
CompactionUtils.getDeltaCommitsSinceLatestCompaction(activeTimeline);
if (deltaCommitsInfoOption.isPresent()) {
Pair<HoodieTimeline, HoodieInstant> deltaCommitsInfo =
deltaCommitsInfoOption.get();
HoodieTimeline deltaCommitTimeline = deltaCommitsInfo.getLeft();
int numDeltaCommits = deltaCommitTimeline.countInstants();
if (numDeltaCommits < maxDeltaCommits) {
- return Option.of(deltaCommitsInfo.getRight());
+ oldestDeltaCommitToRetain = Option.of(deltaCommitsInfo.getRight());
} else {
// delta commits with the last one to keep
List<HoodieInstant> instants =
deltaCommitTimeline.getInstantsAsStream()
.limit(numDeltaCommits - maxDeltaCommits +
1).collect(Collectors.toList());
- return Option.of(instants.get(instants.size() - 1));
+ oldestDeltaCommitToRetain = Option.of(instants.get(instants.size() -
1));
}
}
- return Option.empty();
+
+ HoodieTimeline completedCompactionTimeLine =
activeTimeline.getCommitTimeline()
+ .filterCompletedInstants();
+ Option<HoodieInstant> oldestInstantToRetain =
CleanerUtils.getOldestInstantToRetainFromTimeline(activeTimeline, metaClient,
completedCompactionTimeLine);
+
+ Option<HoodieInstant> finalOldestDeltaCommitToRetain =
oldestDeltaCommitToRetain;
+ if (oldestDeltaCommitToRetain.isPresent()
+ && oldestInstantToRetain.map(instant ->
instant.compareTo(finalOldestDeltaCommitToRetain.get()) > 0).orElse(true)) {
+ oldestInstantToRetain = oldestDeltaCommitToRetain;
+ }
+
+ return oldestInstantToRetain;
Review Comment:
All of these are not necessary.
##########
hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java:
##########
@@ -279,8 +279,8 @@ public static List<HoodieInstant>
getPendingCompactionInstantTimes(HoodieTableMe
*/
public static Option<Pair<HoodieTimeline, HoodieInstant>>
getDeltaCommitsSinceLatestCompaction(
HoodieActiveTimeline activeTimeline) {
- Option<HoodieInstant> lastCompaction = activeTimeline.getCommitTimeline()
- .filterCompletedInstants().lastInstant();
+ Option<HoodieInstant> lastCompaction =
activeTimeline.getCommitAndCompactionTimeline()
+ .lastInstant();
Review Comment:
We should not change this, which is used for
`CompactionTriggerStrategy.NUM_COMMITS`. Otherwise, the semantics would change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@hudi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
