You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2024/05/29 00:24:25 UTC
(hudi) branch master updated: [HUDI-5505] Fix counting of delta commits since last compaction in ScheduleCompactionActionExecutor.getLatestDeltaCommitInfo (#11251)
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 42243862f02 [HUDI-5505] Fix counting of delta commits since last compaction in ScheduleCompactionActionExecutor.getLatestDeltaCommitInfo (#11251)
42243862f02 is described below
commit 42243862f0271fda16e70afdbfde61b47792ff70
Author: Alexander Erofeev <81...@users.noreply.github.com>
AuthorDate: Wed May 29 07:24:20 2024 +0700
[HUDI-5505] Fix counting of delta commits since last compaction in ScheduleCompactionActionExecutor.getLatestDeltaCommitInfo (#11251)
---
.../compact/ScheduleCompactionActionExecutor.java | 2 +-
.../table/action/compact/TestHoodieCompactor.java | 34 +++++++++++++++++++++-
.../apache/hudi/common/util/CompactionUtils.java | 15 ++++++++++
3 files changed, 49 insertions(+), 2 deletions(-)
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
index 941d93fd350..5dc0f067a3d 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/ScheduleCompactionActionExecutor.java
@@ -129,7 +129,7 @@ public class ScheduleCompactionActionExecutor<T, I, K, O> extends BaseActionExec
private Option<Pair<Integer, String>> getLatestDeltaCommitInfo() {
Option<Pair<HoodieTimeline, HoodieInstant>> deltaCommitsInfo =
- CompactionUtils.getDeltaCommitsSinceLatestCompaction(table.getActiveTimeline());
+ CompactionUtils.getCompletedDeltaCommitsSinceLatestCompaction(table.getActiveTimeline());
if (deltaCommitsInfo.isPresent()) {
return Option.of(Pair.of(
deltaCommitsInfo.get().getLeft().countInstants(),
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestHoodieCompactor.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestHoodieCompactor.java
index deda77376af..78086a64359 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestHoodieCompactor.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestHoodieCompactor.java
@@ -89,8 +89,12 @@ public class TestHoodieCompactor extends HoodieSparkClientTestHarness {
}
public HoodieWriteConfig getConfig() {
+ return getConfig(1);
+ }
+
+ public HoodieWriteConfig getConfig(int numCommitsBeforeCompaction) {
return getConfigBuilder()
- .withCompactionConfig(HoodieCompactionConfig.newBuilder().withMaxNumDeltaCommitsBeforeCompaction(1).build())
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder().withMaxNumDeltaCommitsBeforeCompaction(numCommitsBeforeCompaction).build())
.withMetricsConfig(getMetricsConfig())
.build();
}
@@ -181,6 +185,34 @@ public class TestHoodieCompactor extends HoodieSparkClientTestHarness {
}
}
+ @Test
+ public void testNeedCompactionCondition() throws Exception {
+ HoodieWriteConfig config = getConfig(3);
+ try (SparkRDDWriteClient writeClient = getHoodieWriteClient(config)) {
+ // insert 100 records.
+ String newCommitTime = "100";
+ writeClient.startCommitWithTime(newCommitTime);
+
+ // commit 1
+ List<HoodieRecord> records = dataGen.generateInserts(newCommitTime, 100);
+ JavaRDD<HoodieRecord> recordsRDD = jsc.parallelize(records, 1);
+ writeClient.insert(recordsRDD, newCommitTime).collect();
+
+ // commit 2
+ updateRecords(config, "101", records);
+
+ // commit 3 (inflight)
+ newCommitTime = "102";
+ writeClient.startCommitWithTime(newCommitTime);
+ metaClient.getActiveTimeline().transitionRequestedToInflight(new HoodieInstant(State.REQUESTED,
+ HoodieTimeline.DELTA_COMMIT_ACTION, newCommitTime), Option.empty());
+
+ // check that compaction will not be scheduled
+ String compactionTime = "107";
+ assertFalse(writeClient.scheduleCompactionAtInstant(compactionTime, Option.empty()));
+ }
+ }
+
@Test
public void testWriteStatusContentsAfterCompaction() throws Exception {
// insert 100 records
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
index 321fdd0fce5..5c2270db919 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/CompactionUtils.java
@@ -274,6 +274,21 @@ public class CompactionUtils {
return metaClient.getActiveTimeline().filterPendingCompactionTimeline().getInstants();
}
+ /**
+ * Returns a pair of (timeline containing the completed delta commits after the latest completed
+ * compaction commit, the completed compaction commit instant), if the latest completed
+ * compaction commit is present; a pair of (timeline containing all the completed delta commits,
+ * the first delta commit instant), if there is no completed compaction commit.
+ *
+ * @param activeTimeline Active timeline of a table.
+ * @return Pair of timeline containing delta commits and an instant.
+ */
+ public static Option<Pair<HoodieTimeline, HoodieInstant>> getCompletedDeltaCommitsSinceLatestCompaction(
+ HoodieActiveTimeline activeTimeline) {
+ return getDeltaCommitsSinceLatestCompaction(activeTimeline)
+ .map(pair -> Pair.of(pair.getLeft().filterCompletedInstants(), pair.getRight()));
+ }
+
/**
* Returns a pair of (timeline containing the delta commits after the latest completed
* compaction commit, the completed compaction commit instant), if the latest completed