From ddf8fcaf8f6e551fbec41c4e6740dc7e4d925200 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Thu, 7 Jan 2021 14:27:07 +0800 Subject: [PATCH 01/14] update --- .../hudi/config/HoodieCompactionConfig.java | 27 +++++ .../apache/hudi/config/HoodieWriteConfig.java | 18 +++- ...SparkScheduleCompactionActionExecutor.java | 43 ++++++-- .../action/compact/TestInlineCompaction.java | 98 ++++++++++++++++--- 4 files changed, 159 insertions(+), 27 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 08f37740725dd..eabc8bd9aaf6e 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -46,6 +46,9 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { public static final String INLINE_COMPACT_PROP = "hoodie.compact.inline"; // Run a compaction every N delta commits public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_PROP = "hoodie.compact.inline.max.delta.commits"; + public static final String INLINE_COMPACT_ELAPSED_TIME_PROP = "hoodie.compact.inline.max.delta.time"; + public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP = "hoodie.compact.inline.max.delta.commits.enable"; + public static final String INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP = "hoodie.compact.inline.max.delta.time.enable"; public static final String CLEANER_FILE_VERSIONS_RETAINED_PROP = "hoodie.cleaner.fileversions.retained"; public static final String CLEANER_COMMITS_RETAINED_PROP = "hoodie.cleaner.commits.retained"; public static final String CLEANER_INCREMENTAL_MODE = "hoodie.cleaner.incremental.mode"; @@ -109,6 +112,9 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { private static final String DEFAULT_INLINE_COMPACT = "false"; private static final String DEFAULT_INCREMENTAL_CLEANER = "true"; private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS = "5"; + private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME = String.valueOf(60 * 60); + private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED = "true"; + private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME_ENABLED = "false"; private static final String DEFAULT_CLEANER_FILE_VERSIONS_RETAINED = "3"; private static final String DEFAULT_CLEANER_COMMITS_RETAINED = "10"; private static final String DEFAULT_MAX_COMMITS_TO_KEEP = "30"; @@ -235,6 +241,21 @@ public Builder withMaxNumDeltaCommitsBeforeCompaction(int maxNumDeltaCommitsBefo return this; } + public Builder withMaxDeltaTimeBeforeCompaction(int maxDeltaTimeBeforeCompaction) { + props.setProperty(INLINE_COMPACT_ELAPSED_TIME_PROP, String.valueOf(maxDeltaTimeBeforeCompaction)); + return this; + } + + public Builder withMaxNumDeltaCommitsBeforeCompactionEnabled(boolean maxNumDeltaCommitsBeforeCompactionEnabled) { + props.setProperty(INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP, String.valueOf(maxNumDeltaCommitsBeforeCompactionEnabled)); + return this; + } + + public Builder withMaxDeltaTimeBeforeCompactionEnabled(boolean withMaxDeltaTimeBeforeCompactionEnabled) { + props.setProperty(INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP, String.valueOf(withMaxDeltaTimeBeforeCompactionEnabled)); + return this; + } + public Builder withCompactionLazyBlockReadEnabled(Boolean compactionLazyBlockReadEnabled) { props.setProperty(COMPACTION_LAZY_BLOCK_READ_ENABLED_PROP, String.valueOf(compactionLazyBlockReadEnabled)); return this; @@ -271,6 +292,12 @@ public HoodieCompactionConfig build() { DEFAULT_INLINE_COMPACT); setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_NUM_DELTA_COMMITS_PROP), INLINE_COMPACT_NUM_DELTA_COMMITS_PROP, DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS); + setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_ELAPSED_TIME_PROP), + INLINE_COMPACT_ELAPSED_TIME_PROP, DEFAULT_INLINE_COMPACT_ELAPSED_TIME); + setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP), + INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP, DEFAULT_INLINE_COMPACT_ELAPSED_TIME_ENABLED); + setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP), + INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP, DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED); setDefaultOnCondition(props, !props.containsKey(CLEANER_POLICY_PROP), CLEANER_POLICY_PROP, DEFAULT_CLEANER_POLICY); setDefaultOnCondition(props, !props.containsKey(CLEANER_FILE_VERSIONS_RETAINED_PROP), diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 2a26abdb94a0c..d7c257ea340ef 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -18,6 +18,7 @@ package org.apache.hudi.config; +import org.apache.hadoop.hbase.io.compress.Compression; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.bootstrap.BootstrapMode; import org.apache.hudi.common.config.DefaultHoodieConfig; @@ -36,7 +37,6 @@ import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite; import org.apache.hudi.table.action.compact.strategy.CompactionStrategy; -import org.apache.hadoop.hbase.io.compress.Compression; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import javax.annotation.concurrent.Immutable; @@ -397,10 +397,22 @@ public boolean isInlineCompaction() { return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_PROP)); } + public boolean getInlineCompactDeltaElapsedEnabled() { + return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP)); + } + + public boolean getInlineCompactDeltaNumCommitEnabled() { + return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP)); + } + public int getInlineCompactDeltaCommitMax() { return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS_PROP)); } + public int getInlineCompactDeltaElapsedTimeMax() { + return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_ELAPSED_TIME_PROP)); + } + public CompactionStrategy getCompactionStrategy() { return ReflectionUtils.loadClass(props.getProperty(HoodieCompactionConfig.COMPACTION_STRATEGY_PROP)); } @@ -468,7 +480,7 @@ public String getClusteringExecutionStrategyClass() { public long getClusteringMaxBytesInGroup() { return Long.parseLong(props.getProperty(HoodieClusteringConfig.CLUSTERING_MAX_BYTES_PER_GROUP)); } - + public long getClusteringSmallFileLimit() { return Long.parseLong(props.getProperty(HoodieClusteringConfig.CLUSTERING_PLAN_SMALL_FILE_LIMIT)); } @@ -484,7 +496,7 @@ public long getClusteringTargetFileMaxBytes() { public int getTargetPartitionsForClustering() { return Integer.parseInt(props.getProperty(HoodieClusteringConfig.CLUSTERING_TARGET_PARTITIONS)); } - + public String getClusteringSortColumns() { return props.getProperty(HoodieClusteringConfig.CLUSTERING_SORT_COLUMNS_PROPERTY); } diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 96d76d356092a..b101442ded517 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -25,7 +25,9 @@ import org.apache.hudi.common.model.HoodieKey; import org.apache.hudi.common.model.HoodieRecord; import org.apache.hudi.common.model.HoodieRecordPayload; +import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; import org.apache.hudi.common.table.timeline.HoodieInstant; +import org.apache.hudi.common.table.timeline.HoodieTimeline; import org.apache.hudi.common.table.view.SyncableFileSystemView; import org.apache.hudi.common.util.Option; import org.apache.hudi.common.util.collection.Pair; @@ -37,6 +39,7 @@ import org.apache.spark.api.java.JavaRDD; import java.io.IOException; +import java.text.ParseException; import java.util.Map; import java.util.Set; import java.util.stream.Collectors; @@ -60,17 +63,32 @@ protected HoodieCompactionPlan scheduleCompaction() { LOG.info("Checking if compaction needs to be run on " + config.getBasePath()); Option lastCompaction = table.getActiveTimeline().getCommitTimeline() .filterCompletedInstants().lastInstant(); - String lastCompactionTs = "0"; + HoodieTimeline deltaCommits = table.getActiveTimeline().getDeltaCommitTimeline(); + String lastCompactionTs; + int deltaCommitsSinceLastCompaction; if (lastCompaction.isPresent()) { lastCompactionTs = lastCompaction.get().getTimestamp(); + deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfter(lastCompactionTs, Integer.MAX_VALUE).countInstants(); + } else { + lastCompactionTs = deltaCommits.firstInstant().get().getTimestamp(); + deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfterOrEquals(lastCompactionTs, Integer.MAX_VALUE).countInstants(); } - - int deltaCommitsSinceLastCompaction = table.getActiveTimeline().getDeltaCommitTimeline() - .findInstantsAfter(lastCompactionTs, Integer.MAX_VALUE).countInstants(); - if (config.getInlineCompactDeltaCommitMax() > deltaCommitsSinceLastCompaction) { - LOG.info("Not scheduling compaction as only " + deltaCommitsSinceLastCompaction - + " delta commits was found since last compaction " + lastCompactionTs + ". Waiting for " - + config.getInlineCompactDeltaCommitMax()); + // judge if we need to compact according to num delta commits and time elapsed + boolean numCommitEnabled = config.getInlineCompactDeltaNumCommitEnabled(); + boolean timeEnabled = config.getInlineCompactDeltaElapsedEnabled(); + boolean compactable; + if (numCommitEnabled && !timeEnabled) { + compactable = config.getInlineCompactDeltaCommitMax() > deltaCommitsSinceLastCompaction; + } else if (!numCommitEnabled && timeEnabled) { + compactable = parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() > parseToTimestamp(instantTime); + } else { + compactable = config.getInlineCompactDeltaCommitMax() > deltaCommitsSinceLastCompaction + && parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() > parseToTimestamp(instantTime); + } + if (compactable) { + LOG.info(String.format("Not scheduling compaction as only %s delta commits was found since last compaction %s." + + "Waiting for %s,or %sms elapsed time need since last compaction %s.", deltaCommitsSinceLastCompaction, + lastCompactionTs, config.getInlineCompactDeltaCommitMax(), config.getInlineCompactDeltaElapsedTimeMax(), lastCompactionTs)); return new HoodieCompactionPlan(); } @@ -90,4 +108,13 @@ protected HoodieCompactionPlan scheduleCompaction() { } } + public Long parseToTimestamp(String time) { + long timestamp; + try { + timestamp = HoodieActiveTimeline.COMMIT_FORMATTER.parse(time).getTime() / 1000; + } catch (ParseException e) { + throw new HoodieCompactionException(e.getMessage(), e); + } + return timestamp; + } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index 066a9656f2a50..81e91af09b0f1 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -24,7 +24,6 @@ import org.apache.hudi.common.table.HoodieTableMetaClient; import org.apache.hudi.common.table.timeline.HoodieActiveTimeline; import org.apache.hudi.common.table.timeline.HoodieTimeline; -import org.apache.hudi.common.util.CollectionUtils; import org.apache.hudi.config.HoodieCompactionConfig; import org.apache.hudi.config.HoodieWriteConfig; import org.junit.jupiter.api.Test; @@ -39,21 +38,25 @@ public class TestInlineCompaction extends CompactionTestBase { - private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits) { + private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int maxDeltaTime) { return getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() - .withInlineCompaction(true).withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommits).build()) + .withInlineCompaction(true) + .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommits) + .withMaxDeltaTimeBeforeCompaction(maxDeltaTime) + .withMaxDeltaTimeBeforeCompactionEnabled(true).build()) .build(); } @Test public void testCompactionIsNotScheduledEarly() throws Exception { // Given: make two commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { - List records = dataGen.generateInserts("000", 100); + List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 100); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); - runNextDeltaCommits(writeClient, readClient, Arrays.asList("000", "001"), records, cfg, true, new ArrayList<>()); + List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); + runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); // Then: ensure no compaction is executedm since there are only 2 delta commits @@ -62,9 +65,9 @@ public void testCompactionIsNotScheduledEarly() throws Exception { } @Test - public void testSuccessfulCompaction() throws Exception { + public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60); List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { @@ -85,32 +88,95 @@ public void testSuccessfulCompaction() throws Exception { } @Test - public void testCompactionRetryOnFailure() throws Exception { + public void testSuccessfulCompactionBasedOnTime() throws Exception { + // Given: make one commit + HoodieWriteConfig cfg = getConfigForInlineCompaction(5, 10); + + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { + String instantTime = HoodieActiveTimeline.createNewInstantTime(); + List records = dataGen.generateInserts(instantTime, 10); + HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); + runNextDeltaCommits(writeClient, readClient, Arrays.asList(instantTime), records, cfg, true, new ArrayList<>()); + + // after 10s, that will trigger compaction + Thread.sleep(10000); + String finalInstant = HoodieActiveTimeline.createNewInstantTime(); + HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 100), writeClient, metaClient, cfg, false); + + // Then: ensure the file slices are compacted as per policy + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + assertEquals(3, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + assertEquals(HoodieTimeline.COMMIT_ACTION, metaClient.getActiveTimeline().lastInstant().get().getAction()); + } + } + + @Test + public void testCompactionRetryOnFailureBasedOnNumCommits() throws Exception { // Given: two commits, schedule compaction and its failed/in-flight HoodieWriteConfig cfg = getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(false).withMaxNumDeltaCommitsBeforeCompaction(1).build()) .build(); - List instants = CollectionUtils.createImmutableList("000", "001"); + List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); + String instantTime2; try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(instants.get(0), 100); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); - // Schedule compaction 002, make it in-flight (simulates inline compaction failing) - scheduleCompaction("002", writeClient, cfg); - moveCompactionFromRequestedToInflight("002", cfg); + // Schedule compaction instant2, make it in-flight (simulates inline compaction failing) + instantTime2 = HoodieActiveTimeline.createNewInstantTime(); + scheduleCompaction(instantTime2, writeClient, cfg); + moveCompactionFromRequestedToInflight(instantTime2, cfg); } // When: a third commit happens - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60); + String instantTime3 = HoodieActiveTimeline.createNewInstantTime(); + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { + HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + createNextDeltaCommit(instantTime3, dataGen.generateUpdates(instantTime3, 100), writeClient, metaClient, inlineCfg, false); + } + + // Then: 1 delta commit is done, the failed compaction is retried + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + assertEquals(4, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + assertEquals(instantTime2, metaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().firstInstant().get().getTimestamp()); + } + + @Test + public void testCompactionRetryOnFailureBasedOnTime() throws Exception { + // Given: two commits, schedule compaction and its failed/in-flight + HoodieWriteConfig cfg = getConfigBuilder(false) + .withCompactionConfig(HoodieCompactionConfig.newBuilder() + .withInlineCompaction(false).withMaxDeltaTimeBeforeCompaction(5) + .withMaxDeltaTimeBeforeCompactionEnabled(true).build()) + .build(); + String instantTime; + List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { + List records = dataGen.generateInserts(instants.get(0), 100); + HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); + runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); + // Schedule compaction instantTime, make it in-flight (simulates inline compaction failing) + Thread.sleep(10000); + instantTime = HoodieActiveTimeline.createNewInstantTime(); + scheduleCompaction(instantTime, writeClient, cfg); + moveCompactionFromRequestedToInflight(instantTime, cfg); + } + + // When: commit happens after 10s + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(5, 10); + String instantTime2; try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); - createNextDeltaCommit("003", dataGen.generateUpdates("003", 100), writeClient, metaClient, inlineCfg, false); + instantTime2 = HoodieActiveTimeline.createNewInstantTime(); + createNextDeltaCommit(instantTime2, dataGen.generateUpdates(instantTime2, 10), writeClient, metaClient, inlineCfg, false); } // Then: 1 delta commit is done, the failed compaction is retried metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); assertEquals(4, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); - assertEquals("002", metaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().firstInstant().get().getTimestamp()); + assertEquals(instantTime, metaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().firstInstant().get().getTimestamp()); } } From d087f09d4da2b2339413799f4eb13a091faacb9a Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Sun, 10 Jan 2021 14:53:22 +0800 Subject: [PATCH 02/14] update --- .../hudi/config/HoodieCompactionConfig.java | 28 ++--- .../apache/hudi/config/HoodieWriteConfig.java | 9 +- .../table/action/compact/CompactType.java | 23 ++++ ...SparkScheduleCompactionActionExecutor.java | 60 +++++----- .../action/compact/TestInlineCompaction.java | 110 ++++++++++++++++-- .../exception/HoodieCompactException.java | 30 +++++ 6 files changed, 199 insertions(+), 61 deletions(-) create mode 100644 hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java create mode 100644 hudi-common/src/main/java/org/apache/hudi/exception/HoodieCompactException.java diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index eabc8bd9aaf6e..9b91fc8a0ddd1 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -22,6 +22,7 @@ import org.apache.hudi.common.model.HoodieCleaningPolicy; import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload; import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.table.action.compact.CompactType; import org.apache.hudi.table.action.compact.strategy.CompactionStrategy; import org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy; @@ -47,8 +48,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { // Run a compaction every N delta commits public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_PROP = "hoodie.compact.inline.max.delta.commits"; public static final String INLINE_COMPACT_ELAPSED_TIME_PROP = "hoodie.compact.inline.max.delta.time"; - public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP = "hoodie.compact.inline.max.delta.commits.enable"; - public static final String INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP = "hoodie.compact.inline.max.delta.time.enable"; + public static final String INLINE_COMPACT_TYPE_PROP = "hoodie.compact.inline.type"; public static final String CLEANER_FILE_VERSIONS_RETAINED_PROP = "hoodie.cleaner.fileversions.retained"; public static final String CLEANER_COMMITS_RETAINED_PROP = "hoodie.cleaner.commits.retained"; public static final String CLEANER_INCREMENTAL_MODE = "hoodie.cleaner.incremental.mode"; @@ -113,8 +113,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { private static final String DEFAULT_INCREMENTAL_CLEANER = "true"; private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS = "5"; private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME = String.valueOf(60 * 60); - private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED = "true"; - private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME_ENABLED = "false"; + private static final String DEFAULT_INLINE_COMPACT_TYPE = CompactType.COMMIT_NUM.name(); private static final String DEFAULT_CLEANER_FILE_VERSIONS_RETAINED = "3"; private static final String DEFAULT_CLEANER_COMMITS_RETAINED = "10"; private static final String DEFAULT_MAX_COMMITS_TO_KEEP = "30"; @@ -170,6 +169,11 @@ public Builder withInlineCompaction(Boolean inlineCompaction) { return this; } + public Builder withInlineCompactionType(CompactType inlineCompactionType) { + props.setProperty(INLINE_COMPACT_TYPE_PROP, inlineCompactionType.name()); + return this; + } + public Builder withCleanerPolicy(HoodieCleaningPolicy policy) { props.setProperty(CLEANER_POLICY_PROP, policy.name()); return this; @@ -246,16 +250,6 @@ public Builder withMaxDeltaTimeBeforeCompaction(int maxDeltaTimeBeforeCompaction return this; } - public Builder withMaxNumDeltaCommitsBeforeCompactionEnabled(boolean maxNumDeltaCommitsBeforeCompactionEnabled) { - props.setProperty(INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP, String.valueOf(maxNumDeltaCommitsBeforeCompactionEnabled)); - return this; - } - - public Builder withMaxDeltaTimeBeforeCompactionEnabled(boolean withMaxDeltaTimeBeforeCompactionEnabled) { - props.setProperty(INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP, String.valueOf(withMaxDeltaTimeBeforeCompactionEnabled)); - return this; - } - public Builder withCompactionLazyBlockReadEnabled(Boolean compactionLazyBlockReadEnabled) { props.setProperty(COMPACTION_LAZY_BLOCK_READ_ENABLED_PROP, String.valueOf(compactionLazyBlockReadEnabled)); return this; @@ -294,10 +288,8 @@ public HoodieCompactionConfig build() { INLINE_COMPACT_NUM_DELTA_COMMITS_PROP, DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS); setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_ELAPSED_TIME_PROP), INLINE_COMPACT_ELAPSED_TIME_PROP, DEFAULT_INLINE_COMPACT_ELAPSED_TIME); - setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP), - INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP, DEFAULT_INLINE_COMPACT_ELAPSED_TIME_ENABLED); - setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP), - INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP, DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED); + setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_TYPE_PROP), + INLINE_COMPACT_TYPE_PROP, DEFAULT_INLINE_COMPACT_TYPE); setDefaultOnCondition(props, !props.containsKey(CLEANER_POLICY_PROP), CLEANER_POLICY_PROP, DEFAULT_CLEANER_POLICY); setDefaultOnCondition(props, !props.containsKey(CLEANER_FILE_VERSIONS_RETAINED_PROP), diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index d7c257ea340ef..96e60c49d91d8 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -35,6 +35,7 @@ import org.apache.hudi.keygen.SimpleAvroKeyGenerator; import org.apache.hudi.metrics.MetricsReporterType; import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite; +import org.apache.hudi.table.action.compact.CompactType; import org.apache.hudi.table.action.compact.strategy.CompactionStrategy; import org.apache.parquet.hadoop.metadata.CompressionCodecName; @@ -397,12 +398,8 @@ public boolean isInlineCompaction() { return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_PROP)); } - public boolean getInlineCompactDeltaElapsedEnabled() { - return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_ELAPSED_TIME_ENABLED_PROP)); - } - - public boolean getInlineCompactDeltaNumCommitEnabled() { - return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS_ENABLED_PROP)); + public CompactType getInlineCompactType() { + return CompactType.valueOf(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_TYPE_PROP)); } public int getInlineCompactDeltaCommitMax() { diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java new file mode 100644 index 0000000000000..0adf269386e15 --- /dev/null +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java @@ -0,0 +1,23 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.table.action.compact; + +public enum CompactType { + COMMIT_NUM, TIME_ELAPSED, NUM_AND_TIME, NUM_OR_TIME +} diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index b101442ded517..6a601011c88ae 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -74,37 +74,43 @@ protected HoodieCompactionPlan scheduleCompaction() { deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfterOrEquals(lastCompactionTs, Integer.MAX_VALUE).countInstants(); } // judge if we need to compact according to num delta commits and time elapsed - boolean numCommitEnabled = config.getInlineCompactDeltaNumCommitEnabled(); - boolean timeEnabled = config.getInlineCompactDeltaElapsedEnabled(); - boolean compactable; - if (numCommitEnabled && !timeEnabled) { - compactable = config.getInlineCompactDeltaCommitMax() > deltaCommitsSinceLastCompaction; - } else if (!numCommitEnabled && timeEnabled) { - compactable = parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() > parseToTimestamp(instantTime); - } else { - compactable = config.getInlineCompactDeltaCommitMax() > deltaCommitsSinceLastCompaction - && parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() > parseToTimestamp(instantTime); - } + boolean compactable = getCompactType(deltaCommitsSinceLastCompaction, lastCompactionTs); if (compactable) { - LOG.info(String.format("Not scheduling compaction as only %s delta commits was found since last compaction %s." - + "Waiting for %s,or %sms elapsed time need since last compaction %s.", deltaCommitsSinceLastCompaction, - lastCompactionTs, config.getInlineCompactDeltaCommitMax(), config.getInlineCompactDeltaElapsedTimeMax(), lastCompactionTs)); - return new HoodieCompactionPlan(); + LOG.info("Generating compaction plan for merge on read table " + config.getBasePath()); + HoodieSparkMergeOnReadTableCompactor compactor = new HoodieSparkMergeOnReadTableCompactor(); + try { + SyncableFileSystemView fileSystemView = (SyncableFileSystemView) table.getSliceView(); + Set fgInPendingCompactionAndClustering = fileSystemView.getPendingCompactionOperations() + .map(instantTimeOpPair -> instantTimeOpPair.getValue().getFileGroupId()) + .collect(Collectors.toSet()); + // exclude files in pending clustering from compaction. + fgInPendingCompactionAndClustering.addAll(fileSystemView.getFileGroupsInPendingClustering().map(Pair::getLeft).collect(Collectors.toSet())); + return compactor.generateCompactionPlan(context, table, config, instantTime, fgInPendingCompactionAndClustering); + } catch (IOException e) { + throw new HoodieCompactionException("Could not schedule compaction " + config.getBasePath(), e); + } } - LOG.info("Generating compaction plan for merge on read table " + config.getBasePath()); - HoodieSparkMergeOnReadTableCompactor compactor = new HoodieSparkMergeOnReadTableCompactor(); - try { - SyncableFileSystemView fileSystemView = (SyncableFileSystemView) table.getSliceView(); - Set fgInPendingCompactionAndClustering = fileSystemView.getPendingCompactionOperations() - .map(instantTimeOpPair -> instantTimeOpPair.getValue().getFileGroupId()) - .collect(Collectors.toSet()); - // exclude files in pending clustering from compaction. - fgInPendingCompactionAndClustering.addAll(fileSystemView.getFileGroupsInPendingClustering().map(Pair::getLeft).collect(Collectors.toSet())); - return compactor.generateCompactionPlan(context, table, config, instantTime, fgInPendingCompactionAndClustering); + LOG.info(String.format("Not scheduling compaction as only %s delta commits was found since last compaction %s." + + "Waiting for %s,or %sms elapsed time need since last compaction %s.", deltaCommitsSinceLastCompaction, + lastCompactionTs, config.getInlineCompactDeltaCommitMax(), config.getInlineCompactDeltaElapsedTimeMax(), lastCompactionTs)); + return new HoodieCompactionPlan(); + } - } catch (IOException e) { - throw new HoodieCompactionException("Could not schedule compaction " + config.getBasePath(), e); + public boolean getCompactType(Integer deltaCommitsSinceLastCompaction, String lastCompactionTs) { + switch (config.getInlineCompactType()) { + case COMMIT_NUM: + return config.getInlineCompactDeltaCommitMax() <= deltaCommitsSinceLastCompaction; + case TIME_ELAPSED: + return parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + case NUM_OR_TIME: + return config.getInlineCompactDeltaCommitMax() <= deltaCommitsSinceLastCompaction + || parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + case NUM_AND_TIME: + return config.getInlineCompactDeltaCommitMax() <= deltaCommitsSinceLastCompaction + && parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + default: + throw new HoodieCompactionException("Compact type unspecified, set " + config.getInlineCompactType()); } } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index 81e91af09b0f1..337d233e38485 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -38,20 +38,20 @@ public class TestInlineCompaction extends CompactionTestBase { - private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int maxDeltaTime) { + private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int maxDeltaTime, CompactType inlineCompactionType) { return getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(true) .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommits) .withMaxDeltaTimeBeforeCompaction(maxDeltaTime) - .withMaxDeltaTimeBeforeCompactionEnabled(true).build()) + .withInlineCompactionType(inlineCompactionType).build()) .build(); } @Test public void testCompactionIsNotScheduledEarly() throws Exception { // Given: make two commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactType.COMMIT_NUM); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 100); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); @@ -67,7 +67,7 @@ public void testCompactionIsNotScheduledEarly() throws Exception { @Test public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactType.COMMIT_NUM); List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { @@ -90,7 +90,7 @@ public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { @Test public void testSuccessfulCompactionBasedOnTime() throws Exception { // Given: make one commit - HoodieWriteConfig cfg = getConfigForInlineCompaction(5, 10); + HoodieWriteConfig cfg = getConfigForInlineCompaction(5, 10, CompactType.TIME_ELAPSED); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { String instantTime = HoodieActiveTimeline.createNewInstantTime(); @@ -111,12 +111,64 @@ public void testSuccessfulCompactionBasedOnTime() throws Exception { } } + @Test + public void testSuccessfulCompactionBasedOnNumOrTime() throws Exception { + // Given: make three commits + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 20, CompactType.NUM_OR_TIME); + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { + List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 10); + HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); + List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); + runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); + // Then: trigger the compaction because reach 3 commits. + String finalInstant = HoodieActiveTimeline.createNewInstantTime(); + HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 10), writeClient, metaClient, cfg, false); + + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + assertEquals(4, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + Thread.sleep(20000); + // 4th commit, that will trigger compaction because reach the time elapsed + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + finalInstant = HoodieActiveTimeline.createNewInstantTime(); + createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 10), writeClient, metaClient, cfg, false); + + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + assertEquals(6, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + } + } + + @Test + public void testSuccessfulCompactionBasedOnNumAndTime() throws Exception { + // Given: make three commits + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 20, CompactType.NUM_AND_TIME); + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { + List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 10); + HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); + List instants = IntStream.range(0, 3).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); + runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); + HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + + // Then: ensure no compaction is executedm since there are only 3 delta commits + assertEquals(3, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + Thread.sleep(20000); + // 4th commit, that will trigger compaction + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + String finalInstant = HoodieActiveTimeline.createNewInstantTime(); + createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 10), writeClient, metaClient, cfg, false); + + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + assertEquals(5, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + } + } + @Test public void testCompactionRetryOnFailureBasedOnNumCommits() throws Exception { // Given: two commits, schedule compaction and its failed/in-flight HoodieWriteConfig cfg = getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() - .withInlineCompaction(false).withMaxNumDeltaCommitsBeforeCompaction(1).build()) + .withInlineCompaction(false) + .withMaxNumDeltaCommitsBeforeCompaction(1).build()) .build(); List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); String instantTime2; @@ -131,7 +183,7 @@ public void testCompactionRetryOnFailureBasedOnNumCommits() throws Exception { } // When: a third commit happens - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactType.COMMIT_NUM); String instantTime3 = HoodieActiveTimeline.createNewInstantTime(); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); @@ -149,8 +201,9 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { // Given: two commits, schedule compaction and its failed/in-flight HoodieWriteConfig cfg = getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() - .withInlineCompaction(false).withMaxDeltaTimeBeforeCompaction(5) - .withMaxDeltaTimeBeforeCompactionEnabled(true).build()) + .withInlineCompaction(false) + .withMaxDeltaTimeBeforeCompaction(5) + .withInlineCompactionType(CompactType.TIME_ELAPSED).build()) .build(); String instantTime; List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); @@ -166,7 +219,44 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { } // When: commit happens after 10s - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(5, 10); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(5, 10, CompactType.TIME_ELAPSED); + String instantTime2; + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { + HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + instantTime2 = HoodieActiveTimeline.createNewInstantTime(); + createNextDeltaCommit(instantTime2, dataGen.generateUpdates(instantTime2, 10), writeClient, metaClient, inlineCfg, false); + } + + // Then: 1 delta commit is done, the failed compaction is retried + metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); + assertEquals(4, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); + assertEquals(instantTime, metaClient.getActiveTimeline().getCommitTimeline().filterCompletedInstants().firstInstant().get().getTimestamp()); + } + + @Test + public void testCompactionRetryOnFailureBasedOnNumAndTime() throws Exception { + // Given: two commits, schedule compaction and its failed/in-flight + HoodieWriteConfig cfg = getConfigBuilder(false) + .withCompactionConfig(HoodieCompactionConfig.newBuilder() + .withInlineCompaction(false) + .withMaxDeltaTimeBeforeCompaction(1) + .withMaxNumDeltaCommitsBeforeCompaction(1) + .withInlineCompactionType(CompactType.NUM_AND_TIME).build()) + .build(); + String instantTime; + List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); + try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { + List records = dataGen.generateInserts(instants.get(0), 10); + HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); + runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); + // Schedule compaction instantTime, make it in-flight (simulates inline compaction failing) + instantTime = HoodieActiveTimeline.createNewInstantTime(); + scheduleCompaction(instantTime, writeClient, cfg); + moveCompactionFromRequestedToInflight(instantTime, cfg); + } + + // When: a third commit happens + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(3, 20, CompactType.NUM_OR_TIME); String instantTime2; try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); diff --git a/hudi-common/src/main/java/org/apache/hudi/exception/HoodieCompactException.java b/hudi-common/src/main/java/org/apache/hudi/exception/HoodieCompactException.java new file mode 100644 index 0000000000000..0d51706bbec5b --- /dev/null +++ b/hudi-common/src/main/java/org/apache/hudi/exception/HoodieCompactException.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.exception; + +public class HoodieCompactException extends HoodieException { + + public HoodieCompactException(String msg) { + super(msg); + } + + public HoodieCompactException(String msg, Throwable e) { + super(msg, e); + } +} From 7d0453e72c7f9136e16ffaf928e09906917f745f Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Thu, 14 Jan 2021 11:32:57 +0800 Subject: [PATCH 03/14] update --- ...SparkScheduleCompactionActionExecutor.java | 75 ++++++++++++------- 1 file changed, 50 insertions(+), 25 deletions(-) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 6a601011c88ae..67db2dceda84a 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -37,6 +37,7 @@ import org.apache.log4j.LogManager; import org.apache.log4j.Logger; import org.apache.spark.api.java.JavaRDD; +import scala.Tuple2; import java.io.IOException; import java.text.ParseException; @@ -61,20 +62,8 @@ public SparkScheduleCompactionActionExecutor(HoodieEngineContext context, @Override protected HoodieCompactionPlan scheduleCompaction() { LOG.info("Checking if compaction needs to be run on " + config.getBasePath()); - Option lastCompaction = table.getActiveTimeline().getCommitTimeline() - .filterCompletedInstants().lastInstant(); - HoodieTimeline deltaCommits = table.getActiveTimeline().getDeltaCommitTimeline(); - String lastCompactionTs; - int deltaCommitsSinceLastCompaction; - if (lastCompaction.isPresent()) { - lastCompactionTs = lastCompaction.get().getTimestamp(); - deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfter(lastCompactionTs, Integer.MAX_VALUE).countInstants(); - } else { - lastCompactionTs = deltaCommits.firstInstant().get().getTimestamp(); - deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfterOrEquals(lastCompactionTs, Integer.MAX_VALUE).countInstants(); - } // judge if we need to compact according to num delta commits and time elapsed - boolean compactable = getCompactType(deltaCommitsSinceLastCompaction, lastCompactionTs); + boolean compactable = needCompact(config.getInlineCompactType()); if (compactable) { LOG.info("Generating compaction plan for merge on read table " + config.getBasePath()); HoodieSparkMergeOnReadTableCompactor compactor = new HoodieSparkMergeOnReadTableCompactor(); @@ -91,27 +80,63 @@ protected HoodieCompactionPlan scheduleCompaction() { } } - LOG.info(String.format("Not scheduling compaction as only %s delta commits was found since last compaction %s." - + "Waiting for %s,or %sms elapsed time need since last compaction %s.", deltaCommitsSinceLastCompaction, - lastCompactionTs, config.getInlineCompactDeltaCommitMax(), config.getInlineCompactDeltaElapsedTimeMax(), lastCompactionTs)); return new HoodieCompactionPlan(); } - public boolean getCompactType(Integer deltaCommitsSinceLastCompaction, String lastCompactionTs) { - switch (config.getInlineCompactType()) { + public Tuple2 checkCompact(CompactType compactType) { + Option lastCompaction = table.getActiveTimeline().getCommitTimeline() + .filterCompletedInstants().lastInstant(); + HoodieTimeline deltaCommits = table.getActiveTimeline().getDeltaCommitTimeline(); + + String lastCompactionTs; + int deltaCommitsSinceLastCompaction = 0; + if (lastCompaction.isPresent()) { + lastCompactionTs = lastCompaction.get().getTimestamp(); + } else { + lastCompactionTs = deltaCommits.firstInstant().get().getTimestamp(); + } + if (compactType != CompactType.TIME_ELAPSED) { + if (lastCompaction.isPresent()) { + deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfter(lastCompactionTs, Integer.MAX_VALUE).countInstants(); + } else { + deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfterOrEquals(lastCompactionTs, Integer.MAX_VALUE).countInstants(); + } + } + return new Tuple2(deltaCommitsSinceLastCompaction, lastCompactionTs); + } + + public boolean needCompact(CompactType compactType) { + boolean compactable; + // return deltaCommitsSinceLastCompaction and lastCompactionTs + Tuple2 threshold = checkCompact(compactType); + switch (compactType) { case COMMIT_NUM: - return config.getInlineCompactDeltaCommitMax() <= deltaCommitsSinceLastCompaction; + compactable = config.getInlineCompactDeltaCommitMax() <= threshold._1; + break; case TIME_ELAPSED: - return parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + compactable = parseToTimestamp(threshold._2) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + break; case NUM_OR_TIME: - return config.getInlineCompactDeltaCommitMax() <= deltaCommitsSinceLastCompaction - || parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + compactable = config.getInlineCompactDeltaCommitMax() <= threshold._1 + || parseToTimestamp(threshold._2) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + break; case NUM_AND_TIME: - return config.getInlineCompactDeltaCommitMax() <= deltaCommitsSinceLastCompaction - && parseToTimestamp(lastCompactionTs) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + compactable = config.getInlineCompactDeltaCommitMax() <= threshold._1 + && parseToTimestamp(threshold._2) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + break; default: - throw new HoodieCompactionException("Compact type unspecified, set " + config.getInlineCompactType()); + throw new HoodieCompactionException("Unsupported compact type: " + config.getInlineCompactType()); + } + + if (compactable) { + LOG.info(String.format("Scheduling compaction: %s. Delta commits found: %s times, and last compaction time is %s.", + compactType.name(), threshold._1, threshold._2)); + } else { + LOG.info(String.format("Not scheduling compaction as only %s delta commits was found since last compaction %s." + + "Waiting for %s,or %sms elapsed time need since last compaction %s.", threshold._1, + threshold._2, config.getInlineCompactDeltaCommitMax(), config.getInlineCompactDeltaElapsedTimeMax(), threshold._2)); } + return compactable; } public Long parseToTimestamp(String time) { From fbd87b4915f52d3010edeebe66b966855e1abfdc Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Mon, 18 Jan 2021 15:34:35 +0800 Subject: [PATCH 04/14] update --- .../src/main/java/org/apache/hudi/config/HoodieWriteConfig.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 96e60c49d91d8..ea52beced6dac 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -18,7 +18,6 @@ package org.apache.hudi.config; -import org.apache.hadoop.hbase.io.compress.Compression; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.bootstrap.BootstrapMode; import org.apache.hudi.common.config.DefaultHoodieConfig; @@ -38,6 +37,7 @@ import org.apache.hudi.table.action.compact.CompactType; import org.apache.hudi.table.action.compact.strategy.CompactionStrategy; +import org.apache.hadoop.hbase.io.compress.Compression; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import javax.annotation.concurrent.Immutable; From 96e596a8c14dc2a45606a5363e39b87474747b5c Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Mon, 18 Jan 2021 17:56:46 +0800 Subject: [PATCH 05/14] add log in each case --- ...SparkScheduleCompactionActionExecutor.java | 20 +++++++++++++------ 1 file changed, 14 insertions(+), 6 deletions(-) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 67db2dceda84a..e0b951ab3e657 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -109,20 +109,28 @@ public boolean needCompact(CompactType compactType) { boolean compactable; // return deltaCommitsSinceLastCompaction and lastCompactionTs Tuple2 threshold = checkCompact(compactType); + int inlineCompactDeltaCommitMax = config.getInlineCompactDeltaCommitMax(); + int inlineCompactDeltaElapsedTimeMax = config.getInlineCompactDeltaElapsedTimeMax(); switch (compactType) { case COMMIT_NUM: - compactable = config.getInlineCompactDeltaCommitMax() <= threshold._1; + compactable = inlineCompactDeltaCommitMax <= threshold._1; + LOG.info(String.format("Trigger compaction when commit_num >=%s", inlineCompactDeltaCommitMax)); break; case TIME_ELAPSED: - compactable = parseToTimestamp(threshold._2) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + compactable = parseToTimestamp(threshold._2) + inlineCompactDeltaElapsedTimeMax <= parseToTimestamp(instantTime); + LOG.info(String.format("Trigger compaction when elapsed_time >=%ss", inlineCompactDeltaElapsedTimeMax)); break; case NUM_OR_TIME: - compactable = config.getInlineCompactDeltaCommitMax() <= threshold._1 - || parseToTimestamp(threshold._2) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + compactable = inlineCompactDeltaCommitMax <= threshold._1 + || parseToTimestamp(threshold._2) + inlineCompactDeltaElapsedTimeMax <= parseToTimestamp(instantTime); + LOG.info(String.format("Trigger compaction when commit_num >=%s or elapsed_time >=%ss", inlineCompactDeltaCommitMax, + inlineCompactDeltaElapsedTimeMax)); break; case NUM_AND_TIME: - compactable = config.getInlineCompactDeltaCommitMax() <= threshold._1 - && parseToTimestamp(threshold._2) + config.getInlineCompactDeltaElapsedTimeMax() <= parseToTimestamp(instantTime); + compactable = inlineCompactDeltaCommitMax <= threshold._1 + && parseToTimestamp(threshold._2) + inlineCompactDeltaElapsedTimeMax <= parseToTimestamp(instantTime); + LOG.info(String.format("Trigger compaction when commit_num >=%s and elapsed_time >=%ss", inlineCompactDeltaCommitMax, + inlineCompactDeltaElapsedTimeMax)); break; default: throw new HoodieCompactionException("Unsupported compact type: " + config.getInlineCompactType()); From a74ea8120c56e5b40de3491d0a4d61a93981011d Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Tue, 19 Jan 2021 00:32:35 +0800 Subject: [PATCH 06/14] update --- .../hudi/config/HoodieCompactionConfig.java | 6 +- .../apache/hudi/config/HoodieWriteConfig.java | 6 +- ...pe.java => CompactionTriggerStrategy.java} | 4 +- ...SparkScheduleCompactionActionExecutor.java | 80 +++++++++++-------- .../action/compact/TestInlineCompaction.java | 22 ++--- 5 files changed, 66 insertions(+), 52 deletions(-) rename hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/{CompactType.java => CompactionTriggerStrategy.java} (90%) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 9b91fc8a0ddd1..7595b1f9044c0 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -22,7 +22,7 @@ import org.apache.hudi.common.model.HoodieCleaningPolicy; import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload; import org.apache.hudi.common.util.ValidationUtils; -import org.apache.hudi.table.action.compact.CompactType; +import org.apache.hudi.table.action.compact.CompactionTriggerStrategy; import org.apache.hudi.table.action.compact.strategy.CompactionStrategy; import org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy; @@ -113,7 +113,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { private static final String DEFAULT_INCREMENTAL_CLEANER = "true"; private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS = "5"; private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME = String.valueOf(60 * 60); - private static final String DEFAULT_INLINE_COMPACT_TYPE = CompactType.COMMIT_NUM.name(); + private static final String DEFAULT_INLINE_COMPACT_TYPE = CompactionTriggerStrategy.NUM.name(); private static final String DEFAULT_CLEANER_FILE_VERSIONS_RETAINED = "3"; private static final String DEFAULT_CLEANER_COMMITS_RETAINED = "10"; private static final String DEFAULT_MAX_COMMITS_TO_KEEP = "30"; @@ -169,7 +169,7 @@ public Builder withInlineCompaction(Boolean inlineCompaction) { return this; } - public Builder withInlineCompactionType(CompactType inlineCompactionType) { + public Builder withInlineCompactionType(CompactionTriggerStrategy inlineCompactionType) { props.setProperty(INLINE_COMPACT_TYPE_PROP, inlineCompactionType.name()); return this; } diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index ea52beced6dac..5295d13874140 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -34,7 +34,7 @@ import org.apache.hudi.keygen.SimpleAvroKeyGenerator; import org.apache.hudi.metrics.MetricsReporterType; import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite; -import org.apache.hudi.table.action.compact.CompactType; +import org.apache.hudi.table.action.compact.CompactionTriggerStrategy; import org.apache.hudi.table.action.compact.strategy.CompactionStrategy; import org.apache.hadoop.hbase.io.compress.Compression; @@ -398,8 +398,8 @@ public boolean isInlineCompaction() { return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_PROP)); } - public CompactType getInlineCompactType() { - return CompactType.valueOf(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_TYPE_PROP)); + public CompactionTriggerStrategy getInlineCompactType() { + return CompactionTriggerStrategy.valueOf(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_TYPE_PROP)); } public int getInlineCompactDeltaCommitMax() { diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java similarity index 90% rename from hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java rename to hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java index 0adf269386e15..faba023b01472 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactType.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java @@ -18,6 +18,6 @@ package org.apache.hudi.table.action.compact; -public enum CompactType { - COMMIT_NUM, TIME_ELAPSED, NUM_AND_TIME, NUM_OR_TIME +public enum CompactionTriggerStrategy { + NUM, TIME_ELAPSED, NUM_AND_TIME, NUM_OR_TIME } diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index e0b951ab3e657..46bb000ff1f26 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -83,9 +83,9 @@ protected HoodieCompactionPlan scheduleCompaction() { return new HoodieCompactionPlan(); } - public Tuple2 checkCompact(CompactType compactType) { + public Tuple2 getLastDeltaCommitInfo(CompactionTriggerStrategy compactionTriggerStrategy) { Option lastCompaction = table.getActiveTimeline().getCommitTimeline() - .filterCompletedInstants().lastInstant(); + .filterCompletedInstants().lastInstant(); HoodieTimeline deltaCommits = table.getActiveTimeline().getDeltaCommitTimeline(); String lastCompactionTs; @@ -95,7 +95,7 @@ public Tuple2 checkCompact(CompactType compactType) { } else { lastCompactionTs = deltaCommits.firstInstant().get().getTimestamp(); } - if (compactType != CompactType.TIME_ELAPSED) { + if (compactionTriggerStrategy != CompactionTriggerStrategy.TIME_ELAPSED) { if (lastCompaction.isPresent()) { deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfter(lastCompactionTs, Integer.MAX_VALUE).countInstants(); } else { @@ -105,49 +105,63 @@ public Tuple2 checkCompact(CompactType compactType) { return new Tuple2(deltaCommitsSinceLastCompaction, lastCompactionTs); } - public boolean needCompact(CompactType compactType) { + public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) { boolean compactable; // return deltaCommitsSinceLastCompaction and lastCompactionTs - Tuple2 threshold = checkCompact(compactType); + Tuple2 threshold = getLastDeltaCommitInfo(compactionTriggerStrategy); int inlineCompactDeltaCommitMax = config.getInlineCompactDeltaCommitMax(); int inlineCompactDeltaElapsedTimeMax = config.getInlineCompactDeltaElapsedTimeMax(); - switch (compactType) { - case COMMIT_NUM: + long elapsedTime; + switch (compactionTriggerStrategy) { + case NUM: compactable = inlineCompactDeltaCommitMax <= threshold._1; - LOG.info(String.format("Trigger compaction when commit_num >=%s", inlineCompactDeltaCommitMax)); - break; + if (compactable) { + LOG.info(String.format("The delta commits >= %s, trigger compaction scheduler.", inlineCompactDeltaCommitMax)); + } else { + LOG.info(String.format("Not scheduling compaction because %s delta commits needed since last compaction %s." + + "But only %s delta commits found.", inlineCompactDeltaCommitMax, threshold._2, threshold._1)); + } + return compactable; case TIME_ELAPSED: - compactable = parseToTimestamp(threshold._2) + inlineCompactDeltaElapsedTimeMax <= parseToTimestamp(instantTime); - LOG.info(String.format("Trigger compaction when elapsed_time >=%ss", inlineCompactDeltaElapsedTimeMax)); - break; + elapsedTime = parsedToSeconds(instantTime) - parsedToSeconds(threshold._2); + compactable = inlineCompactDeltaElapsedTimeMax <= elapsedTime; + if (compactable) { + LOG.info(String.format("The elapsed time >=%ss, trigger compaction scheduler.", inlineCompactDeltaElapsedTimeMax)); + } else { + LOG.info(String.format("Not scheduling compaction because %s elapsed time needed since last compaction %s." + + "But only %ss elapsed time found", inlineCompactDeltaElapsedTimeMax, threshold._2, elapsedTime)); + } + return compactable; case NUM_OR_TIME: - compactable = inlineCompactDeltaCommitMax <= threshold._1 - || parseToTimestamp(threshold._2) + inlineCompactDeltaElapsedTimeMax <= parseToTimestamp(instantTime); - LOG.info(String.format("Trigger compaction when commit_num >=%s or elapsed_time >=%ss", inlineCompactDeltaCommitMax, - inlineCompactDeltaElapsedTimeMax)); - break; + elapsedTime = parsedToSeconds(instantTime) - parsedToSeconds(threshold._2); + compactable = inlineCompactDeltaCommitMax <= threshold._1 || inlineCompactDeltaElapsedTimeMax <= elapsedTime; + if (compactable) { + LOG.info(String.format("The delta commits >= %s or elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, + inlineCompactDeltaElapsedTimeMax)); + } else { + LOG.info(String.format("Not scheduling compaction because %s delta commits or %ss elapsed time needed since last compaction %s." + + "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, + threshold._1, elapsedTime)); + } + return compactable; case NUM_AND_TIME: - compactable = inlineCompactDeltaCommitMax <= threshold._1 - && parseToTimestamp(threshold._2) + inlineCompactDeltaElapsedTimeMax <= parseToTimestamp(instantTime); - LOG.info(String.format("Trigger compaction when commit_num >=%s and elapsed_time >=%ss", inlineCompactDeltaCommitMax, - inlineCompactDeltaElapsedTimeMax)); - break; + elapsedTime = parsedToSeconds(instantTime) - parsedToSeconds(threshold._2); + compactable = inlineCompactDeltaCommitMax <= threshold._1 && inlineCompactDeltaElapsedTimeMax <= elapsedTime; + if (compactable) { + LOG.info(String.format("The delta commits >= %s and elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, + inlineCompactDeltaElapsedTimeMax)); + } else { + LOG.info(String.format("Not scheduling compaction because %s delta commits and %ss elapsed time needed since last compaction %s." + + "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, + threshold._1, elapsedTime)); + } + return compactable; default: throw new HoodieCompactionException("Unsupported compact type: " + config.getInlineCompactType()); } - - if (compactable) { - LOG.info(String.format("Scheduling compaction: %s. Delta commits found: %s times, and last compaction time is %s.", - compactType.name(), threshold._1, threshold._2)); - } else { - LOG.info(String.format("Not scheduling compaction as only %s delta commits was found since last compaction %s." - + "Waiting for %s,or %sms elapsed time need since last compaction %s.", threshold._1, - threshold._2, config.getInlineCompactDeltaCommitMax(), config.getInlineCompactDeltaElapsedTimeMax(), threshold._2)); - } - return compactable; } - public Long parseToTimestamp(String time) { + public Long parsedToSeconds(String time) { long timestamp; try { timestamp = HoodieActiveTimeline.COMMIT_FORMATTER.parse(time).getTime() / 1000; diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index 337d233e38485..ca5b1d443dec9 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -38,7 +38,7 @@ public class TestInlineCompaction extends CompactionTestBase { - private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int maxDeltaTime, CompactType inlineCompactionType) { + private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int maxDeltaTime, CompactionTriggerStrategy inlineCompactionType) { return getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(true) @@ -51,7 +51,7 @@ private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int @Test public void testCompactionIsNotScheduledEarly() throws Exception { // Given: make two commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactType.COMMIT_NUM); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.COMMIT_NUM); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 100); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); @@ -67,7 +67,7 @@ public void testCompactionIsNotScheduledEarly() throws Exception { @Test public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactType.COMMIT_NUM); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.COMMIT_NUM); List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { @@ -90,7 +90,7 @@ public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { @Test public void testSuccessfulCompactionBasedOnTime() throws Exception { // Given: make one commit - HoodieWriteConfig cfg = getConfigForInlineCompaction(5, 10, CompactType.TIME_ELAPSED); + HoodieWriteConfig cfg = getConfigForInlineCompaction(5, 10, CompactionTriggerStrategy.TIME_ELAPSED); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { String instantTime = HoodieActiveTimeline.createNewInstantTime(); @@ -114,7 +114,7 @@ public void testSuccessfulCompactionBasedOnTime() throws Exception { @Test public void testSuccessfulCompactionBasedOnNumOrTime() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 20, CompactType.NUM_OR_TIME); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 20, CompactionTriggerStrategy.NUM_OR_TIME); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 10); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); @@ -141,7 +141,7 @@ public void testSuccessfulCompactionBasedOnNumOrTime() throws Exception { @Test public void testSuccessfulCompactionBasedOnNumAndTime() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 20, CompactType.NUM_AND_TIME); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 20, CompactionTriggerStrategy.NUM_AND_TIME); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 10); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); @@ -183,7 +183,7 @@ public void testCompactionRetryOnFailureBasedOnNumCommits() throws Exception { } // When: a third commit happens - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactType.COMMIT_NUM); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactionTriggerStrategy.COMMIT_NUM); String instantTime3 = HoodieActiveTimeline.createNewInstantTime(); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); @@ -203,7 +203,7 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(false) .withMaxDeltaTimeBeforeCompaction(5) - .withInlineCompactionType(CompactType.TIME_ELAPSED).build()) + .withInlineCompactionType(CompactionTriggerStrategy.TIME_ELAPSED).build()) .build(); String instantTime; List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); @@ -219,7 +219,7 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { } // When: commit happens after 10s - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(5, 10, CompactType.TIME_ELAPSED); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(5, 10, CompactionTriggerStrategy.TIME_ELAPSED); String instantTime2; try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); @@ -241,7 +241,7 @@ public void testCompactionRetryOnFailureBasedOnNumAndTime() throws Exception { .withInlineCompaction(false) .withMaxDeltaTimeBeforeCompaction(1) .withMaxNumDeltaCommitsBeforeCompaction(1) - .withInlineCompactionType(CompactType.NUM_AND_TIME).build()) + .withInlineCompactionType(CompactionTriggerStrategy.NUM_AND_TIME).build()) .build(); String instantTime; List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); @@ -256,7 +256,7 @@ public void testCompactionRetryOnFailureBasedOnNumAndTime() throws Exception { } // When: a third commit happens - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(3, 20, CompactType.NUM_OR_TIME); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(3, 20, CompactionTriggerStrategy.NUM_OR_TIME); String instantTime2; try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); From 4259478b945ea0440e0c28c56dc6290ee2df4442 Mon Sep 17 00:00:00 2001 From: Karl_Wang Date: Tue, 19 Jan 2021 00:44:59 +0800 Subject: [PATCH 07/14] Update HoodieWriteConfig.java --- .../main/java/org/apache/hudi/config/HoodieWriteConfig.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 5295d13874140..ccf53a56a4206 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -477,7 +477,7 @@ public String getClusteringExecutionStrategyClass() { public long getClusteringMaxBytesInGroup() { return Long.parseLong(props.getProperty(HoodieClusteringConfig.CLUSTERING_MAX_BYTES_PER_GROUP)); } - + public long getClusteringSmallFileLimit() { return Long.parseLong(props.getProperty(HoodieClusteringConfig.CLUSTERING_PLAN_SMALL_FILE_LIMIT)); } @@ -493,7 +493,7 @@ public long getClusteringTargetFileMaxBytes() { public int getTargetPartitionsForClustering() { return Integer.parseInt(props.getProperty(HoodieClusteringConfig.CLUSTERING_TARGET_PARTITIONS)); } - + public String getClusteringSortColumns() { return props.getProperty(HoodieClusteringConfig.CLUSTERING_SORT_COLUMNS_PROPERTY); } From 3c3e12ef34026b9d5b89d3d537a37040046abae3 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Tue, 19 Jan 2021 01:02:14 +0800 Subject: [PATCH 08/14] update --- .../apache/hudi/config/HoodieCompactionConfig.java | 12 ++++++------ .../org/apache/hudi/config/HoodieWriteConfig.java | 4 ++-- .../SparkScheduleCompactionActionExecutor.java | 4 ++-- 3 files changed, 10 insertions(+), 10 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 7595b1f9044c0..c6335b702b98a 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -48,7 +48,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { // Run a compaction every N delta commits public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_PROP = "hoodie.compact.inline.max.delta.commits"; public static final String INLINE_COMPACT_ELAPSED_TIME_PROP = "hoodie.compact.inline.max.delta.time"; - public static final String INLINE_COMPACT_TYPE_PROP = "hoodie.compact.inline.type"; + public static final String INLINE_COMPACT_TRIGGER_STRATEGY_PROP = "hoodie.compact.inline.trigger.strategy"; public static final String CLEANER_FILE_VERSIONS_RETAINED_PROP = "hoodie.cleaner.fileversions.retained"; public static final String CLEANER_COMMITS_RETAINED_PROP = "hoodie.cleaner.commits.retained"; public static final String CLEANER_INCREMENTAL_MODE = "hoodie.cleaner.incremental.mode"; @@ -113,7 +113,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { private static final String DEFAULT_INCREMENTAL_CLEANER = "true"; private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS = "5"; private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME = String.valueOf(60 * 60); - private static final String DEFAULT_INLINE_COMPACT_TYPE = CompactionTriggerStrategy.NUM.name(); + private static final String DEFAULT_INLINE_COMPACT_TRIGGER_STRATEGY = CompactionTriggerStrategy.NUM.name(); private static final String DEFAULT_CLEANER_FILE_VERSIONS_RETAINED = "3"; private static final String DEFAULT_CLEANER_COMMITS_RETAINED = "10"; private static final String DEFAULT_MAX_COMMITS_TO_KEEP = "30"; @@ -169,8 +169,8 @@ public Builder withInlineCompaction(Boolean inlineCompaction) { return this; } - public Builder withInlineCompactionType(CompactionTriggerStrategy inlineCompactionType) { - props.setProperty(INLINE_COMPACT_TYPE_PROP, inlineCompactionType.name()); + public Builder withInlineCompactionType(CompactionTriggerStrategy compactionTriggerStrategy) { + props.setProperty(INLINE_COMPACT_TRIGGER_STRATEGY_PROP, compactionTriggerStrategy.name()); return this; } @@ -288,8 +288,8 @@ public HoodieCompactionConfig build() { INLINE_COMPACT_NUM_DELTA_COMMITS_PROP, DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS); setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_ELAPSED_TIME_PROP), INLINE_COMPACT_ELAPSED_TIME_PROP, DEFAULT_INLINE_COMPACT_ELAPSED_TIME); - setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_TYPE_PROP), - INLINE_COMPACT_TYPE_PROP, DEFAULT_INLINE_COMPACT_TYPE); + setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_TRIGGER_STRATEGY_PROP), + INLINE_COMPACT_TRIGGER_STRATEGY_PROP, DEFAULT_INLINE_COMPACT_TRIGGER_STRATEGY); setDefaultOnCondition(props, !props.containsKey(CLEANER_POLICY_PROP), CLEANER_POLICY_PROP, DEFAULT_CLEANER_POLICY); setDefaultOnCondition(props, !props.containsKey(CLEANER_FILE_VERSIONS_RETAINED_PROP), diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 5295d13874140..b1155e3bc9d1f 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -398,8 +398,8 @@ public boolean isInlineCompaction() { return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_PROP)); } - public CompactionTriggerStrategy getInlineCompactType() { - return CompactionTriggerStrategy.valueOf(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_TYPE_PROP)); + public CompactionTriggerStrategy getInlineCompactTriggerStrategy() { + return CompactionTriggerStrategy.valueOf(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_TRIGGER_STRATEGY_PROP)); } public int getInlineCompactDeltaCommitMax() { diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 46bb000ff1f26..b7d4d58334fe0 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -63,7 +63,7 @@ public SparkScheduleCompactionActionExecutor(HoodieEngineContext context, protected HoodieCompactionPlan scheduleCompaction() { LOG.info("Checking if compaction needs to be run on " + config.getBasePath()); // judge if we need to compact according to num delta commits and time elapsed - boolean compactable = needCompact(config.getInlineCompactType()); + boolean compactable = needCompact(config.getInlineCompactTriggerStrategy()); if (compactable) { LOG.info("Generating compaction plan for merge on read table " + config.getBasePath()); HoodieSparkMergeOnReadTableCompactor compactor = new HoodieSparkMergeOnReadTableCompactor(); @@ -157,7 +157,7 @@ public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) } return compactable; default: - throw new HoodieCompactionException("Unsupported compact type: " + config.getInlineCompactType()); + throw new HoodieCompactionException("Unsupported compact type: " + config.getInlineCompactTriggerStrategy()); } } From b9a1a63910825d68b9a7e99cd9dd9995a4312448 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Tue, 19 Jan 2021 09:49:36 +0800 Subject: [PATCH 09/14] update --- .../SparkScheduleCompactionActionExecutor.java | 16 ++++++++-------- .../action/compact/TestInlineCompaction.java | 6 +++--- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index b7d4d58334fe0..8492b229a30f7 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -118,8 +118,8 @@ public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) if (compactable) { LOG.info(String.format("The delta commits >= %s, trigger compaction scheduler.", inlineCompactDeltaCommitMax)); } else { - LOG.info(String.format("Not scheduling compaction because %s delta commits needed since last compaction %s." + - "But only %s delta commits found.", inlineCompactDeltaCommitMax, threshold._2, threshold._1)); + LOG.info(String.format("Not scheduling compaction because %s delta commits needed since last compaction %s." + + "But only %s delta commits found.", inlineCompactDeltaCommitMax, threshold._2, threshold._1)); } return compactable; case TIME_ELAPSED: @@ -128,8 +128,8 @@ public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) if (compactable) { LOG.info(String.format("The elapsed time >=%ss, trigger compaction scheduler.", inlineCompactDeltaElapsedTimeMax)); } else { - LOG.info(String.format("Not scheduling compaction because %s elapsed time needed since last compaction %s." + - "But only %ss elapsed time found", inlineCompactDeltaElapsedTimeMax, threshold._2, elapsedTime)); + LOG.info(String.format("Not scheduling compaction because %s elapsed time needed since last compaction %s." + + "But only %ss elapsed time found", inlineCompactDeltaElapsedTimeMax, threshold._2, elapsedTime)); } return compactable; case NUM_OR_TIME: @@ -139,8 +139,8 @@ public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) LOG.info(String.format("The delta commits >= %s or elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax)); } else { - LOG.info(String.format("Not scheduling compaction because %s delta commits or %ss elapsed time needed since last compaction %s." + - "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, + LOG.info(String.format("Not scheduling compaction because %s delta commits or %ss elapsed time needed since last compaction %s." + + "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, threshold._1, elapsedTime)); } return compactable; @@ -151,8 +151,8 @@ public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) LOG.info(String.format("The delta commits >= %s and elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax)); } else { - LOG.info(String.format("Not scheduling compaction because %s delta commits and %ss elapsed time needed since last compaction %s." + - "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, + LOG.info(String.format("Not scheduling compaction because %s delta commits and %ss elapsed time needed since last compaction %s." + + "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, threshold._1, elapsedTime)); } return compactable; diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index ca5b1d443dec9..0fe993232d90b 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -51,7 +51,7 @@ private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int @Test public void testCompactionIsNotScheduledEarly() throws Exception { // Given: make two commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.COMMIT_NUM); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.NUM); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 100); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); @@ -67,7 +67,7 @@ public void testCompactionIsNotScheduledEarly() throws Exception { @Test public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.COMMIT_NUM); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.NUM); List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { @@ -183,7 +183,7 @@ public void testCompactionRetryOnFailureBasedOnNumCommits() throws Exception { } // When: a third commit happens - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactionTriggerStrategy.COMMIT_NUM); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactionTriggerStrategy.NUM); String instantTime3 = HoodieActiveTimeline.createNewInstantTime(); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); From d720b4e939dab49e8493dbce073c7ee859b03594 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Tue, 19 Jan 2021 09:53:41 +0800 Subject: [PATCH 10/14] update --- .../java/org/apache/hudi/config/HoodieCompactionConfig.java | 2 +- .../hudi/table/action/compact/TestInlineCompaction.java | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index c6335b702b98a..2710b836a5262 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -169,7 +169,7 @@ public Builder withInlineCompaction(Boolean inlineCompaction) { return this; } - public Builder withInlineCompactionType(CompactionTriggerStrategy compactionTriggerStrategy) { + public Builder withInlineCompactionTriggerStrategy(CompactionTriggerStrategy compactionTriggerStrategy) { props.setProperty(INLINE_COMPACT_TRIGGER_STRATEGY_PROP, compactionTriggerStrategy.name()); return this; } diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index 0fe993232d90b..1e7b37272a2cc 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -44,7 +44,7 @@ private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int .withInlineCompaction(true) .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommits) .withMaxDeltaTimeBeforeCompaction(maxDeltaTime) - .withInlineCompactionType(inlineCompactionType).build()) + .withInlineCompactionTriggerStrategy(inlineCompactionType).build()) .build(); } @@ -203,7 +203,7 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(false) .withMaxDeltaTimeBeforeCompaction(5) - .withInlineCompactionType(CompactionTriggerStrategy.TIME_ELAPSED).build()) + .withInlineCompactionTriggerStrategy(CompactionTriggerStrategy.TIME_ELAPSED).build()) .build(); String instantTime; List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); @@ -241,7 +241,7 @@ public void testCompactionRetryOnFailureBasedOnNumAndTime() throws Exception { .withInlineCompaction(false) .withMaxDeltaTimeBeforeCompaction(1) .withMaxNumDeltaCommitsBeforeCompaction(1) - .withInlineCompactionType(CompactionTriggerStrategy.NUM_AND_TIME).build()) + .withInlineCompactionTriggerStrategy(CompactionTriggerStrategy.NUM_AND_TIME).build()) .build(); String instantTime; List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); From 48f7392acd2021a8cbfc3a39e71189a9564e2c4f Mon Sep 17 00:00:00 2001 From: Karl_Wang Date: Tue, 19 Jan 2021 22:38:31 +0800 Subject: [PATCH 11/14] Update HoodieCompactionConfig.java --- .../java/org/apache/hudi/config/HoodieCompactionConfig.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 2710b836a5262..9d43733d29a4c 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -47,7 +47,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { public static final String INLINE_COMPACT_PROP = "hoodie.compact.inline"; // Run a compaction every N delta commits public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_PROP = "hoodie.compact.inline.max.delta.commits"; - public static final String INLINE_COMPACT_ELAPSED_TIME_PROP = "hoodie.compact.inline.max.delta.time"; + public static final String INLINE_COMPACT_ELAPSED_TIME_PROP = "hoodie.compact.inline.max.delta.seconds"; public static final String INLINE_COMPACT_TRIGGER_STRATEGY_PROP = "hoodie.compact.inline.trigger.strategy"; public static final String CLEANER_FILE_VERSIONS_RETAINED_PROP = "hoodie.cleaner.fileversions.retained"; public static final String CLEANER_COMMITS_RETAINED_PROP = "hoodie.cleaner.commits.retained"; From c2a695a7fc90389ed68bedbd0677bea8820e47a0 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Wed, 20 Jan 2021 22:44:25 +0800 Subject: [PATCH 12/14] update --- .../hudi/config/HoodieCompactionConfig.java | 12 ++--- .../apache/hudi/config/HoodieWriteConfig.java | 4 +- ...SparkScheduleCompactionActionExecutor.java | 51 +++++++------------ .../action/compact/TestInlineCompaction.java | 18 +++---- .../table/timeline/HoodieActiveTimeline.java | 14 +++++ 5 files changed, 47 insertions(+), 52 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 9d43733d29a4c..00b8d5eec73d1 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -47,7 +47,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { public static final String INLINE_COMPACT_PROP = "hoodie.compact.inline"; // Run a compaction every N delta commits public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_PROP = "hoodie.compact.inline.max.delta.commits"; - public static final String INLINE_COMPACT_ELAPSED_TIME_PROP = "hoodie.compact.inline.max.delta.seconds"; + public static final String INLINE_COMPACT_TIME_DELTA_SECONDS_PROP = "hoodie.compact.inline.max.delta.seconds"; public static final String INLINE_COMPACT_TRIGGER_STRATEGY_PROP = "hoodie.compact.inline.trigger.strategy"; public static final String CLEANER_FILE_VERSIONS_RETAINED_PROP = "hoodie.cleaner.fileversions.retained"; public static final String CLEANER_COMMITS_RETAINED_PROP = "hoodie.cleaner.commits.retained"; @@ -112,7 +112,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { private static final String DEFAULT_INLINE_COMPACT = "false"; private static final String DEFAULT_INCREMENTAL_CLEANER = "true"; private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS = "5"; - private static final String DEFAULT_INLINE_COMPACT_ELAPSED_TIME = String.valueOf(60 * 60); + private static final String DEFAULT_INLINE_COMPACT_TIME_DELTA_SECONDS = String.valueOf(60 * 60); private static final String DEFAULT_INLINE_COMPACT_TRIGGER_STRATEGY = CompactionTriggerStrategy.NUM.name(); private static final String DEFAULT_CLEANER_FILE_VERSIONS_RETAINED = "3"; private static final String DEFAULT_CLEANER_COMMITS_RETAINED = "10"; @@ -245,8 +245,8 @@ public Builder withMaxNumDeltaCommitsBeforeCompaction(int maxNumDeltaCommitsBefo return this; } - public Builder withMaxDeltaTimeBeforeCompaction(int maxDeltaTimeBeforeCompaction) { - props.setProperty(INLINE_COMPACT_ELAPSED_TIME_PROP, String.valueOf(maxDeltaTimeBeforeCompaction)); + public Builder withMaxDeltaSecondsBeforeCompaction(int maxDeltaSecondsBeforeCompaction) { + props.setProperty(INLINE_COMPACT_TIME_DELTA_SECONDS_PROP, String.valueOf(maxDeltaSecondsBeforeCompaction)); return this; } @@ -286,8 +286,8 @@ public HoodieCompactionConfig build() { DEFAULT_INLINE_COMPACT); setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_NUM_DELTA_COMMITS_PROP), INLINE_COMPACT_NUM_DELTA_COMMITS_PROP, DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS); - setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_ELAPSED_TIME_PROP), - INLINE_COMPACT_ELAPSED_TIME_PROP, DEFAULT_INLINE_COMPACT_ELAPSED_TIME); + setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_TIME_DELTA_SECONDS_PROP), + INLINE_COMPACT_TIME_DELTA_SECONDS_PROP, DEFAULT_INLINE_COMPACT_TIME_DELTA_SECONDS); setDefaultOnCondition(props, !props.containsKey(INLINE_COMPACT_TRIGGER_STRATEGY_PROP), INLINE_COMPACT_TRIGGER_STRATEGY_PROP, DEFAULT_INLINE_COMPACT_TRIGGER_STRATEGY); setDefaultOnCondition(props, !props.containsKey(CLEANER_POLICY_PROP), CLEANER_POLICY_PROP, diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java index 251ea2305f324..4b54219348af6 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java @@ -406,8 +406,8 @@ public int getInlineCompactDeltaCommitMax() { return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS_PROP)); } - public int getInlineCompactDeltaElapsedTimeMax() { - return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_ELAPSED_TIME_PROP)); + public int getInlineCompactDeltaSecondsMax() { + return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_TIME_DELTA_SECONDS_PROP)); } public CompactionStrategy getCompactionStrategy() { diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 8492b229a30f7..1ec867cb39e58 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -107,58 +107,43 @@ public Tuple2 getLastDeltaCommitInfo(CompactionTriggerStrategy public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) { boolean compactable; - // return deltaCommitsSinceLastCompaction and lastCompactionTs - Tuple2 threshold = getLastDeltaCommitInfo(compactionTriggerStrategy); + // get deltaCommitsSinceLastCompaction and lastCompactionTs + Tuple2 lastDeltaCommitInfo = getLastDeltaCommitInfo(compactionTriggerStrategy); int inlineCompactDeltaCommitMax = config.getInlineCompactDeltaCommitMax(); - int inlineCompactDeltaElapsedTimeMax = config.getInlineCompactDeltaElapsedTimeMax(); - long elapsedTime; + int inlineCompactDeltaSecondsMax = config.getInlineCompactDeltaSecondsMax(); switch (compactionTriggerStrategy) { case NUM: - compactable = inlineCompactDeltaCommitMax <= threshold._1; + compactable = inlineCompactDeltaCommitMax <= lastDeltaCommitInfo._1; if (compactable) { LOG.info(String.format("The delta commits >= %s, trigger compaction scheduler.", inlineCompactDeltaCommitMax)); - } else { - LOG.info(String.format("Not scheduling compaction because %s delta commits needed since last compaction %s." - + "But only %s delta commits found.", inlineCompactDeltaCommitMax, threshold._2, threshold._1)); } - return compactable; + break; case TIME_ELAPSED: - elapsedTime = parsedToSeconds(instantTime) - parsedToSeconds(threshold._2); - compactable = inlineCompactDeltaElapsedTimeMax <= elapsedTime; + compactable = inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(lastDeltaCommitInfo._2); if (compactable) { - LOG.info(String.format("The elapsed time >=%ss, trigger compaction scheduler.", inlineCompactDeltaElapsedTimeMax)); - } else { - LOG.info(String.format("Not scheduling compaction because %s elapsed time needed since last compaction %s." - + "But only %ss elapsed time found", inlineCompactDeltaElapsedTimeMax, threshold._2, elapsedTime)); + LOG.info(String.format("The elapsed time >=%ss, trigger compaction scheduler.", inlineCompactDeltaSecondsMax)); } - return compactable; + break; case NUM_OR_TIME: - elapsedTime = parsedToSeconds(instantTime) - parsedToSeconds(threshold._2); - compactable = inlineCompactDeltaCommitMax <= threshold._1 || inlineCompactDeltaElapsedTimeMax <= elapsedTime; + compactable = inlineCompactDeltaCommitMax <= lastDeltaCommitInfo._1 + || inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(lastDeltaCommitInfo._2); if (compactable) { LOG.info(String.format("The delta commits >= %s or elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, - inlineCompactDeltaElapsedTimeMax)); - } else { - LOG.info(String.format("Not scheduling compaction because %s delta commits or %ss elapsed time needed since last compaction %s." - + "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, - threshold._1, elapsedTime)); + inlineCompactDeltaSecondsMax)); } - return compactable; + break; case NUM_AND_TIME: - elapsedTime = parsedToSeconds(instantTime) - parsedToSeconds(threshold._2); - compactable = inlineCompactDeltaCommitMax <= threshold._1 && inlineCompactDeltaElapsedTimeMax <= elapsedTime; + compactable = inlineCompactDeltaCommitMax <= lastDeltaCommitInfo._1 + && inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(lastDeltaCommitInfo._2); if (compactable) { LOG.info(String.format("The delta commits >= %s and elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, - inlineCompactDeltaElapsedTimeMax)); - } else { - LOG.info(String.format("Not scheduling compaction because %s delta commits and %ss elapsed time needed since last compaction %s." - + "But only %s delta commits and %ss elapsed time found", inlineCompactDeltaCommitMax, inlineCompactDeltaElapsedTimeMax, threshold._2, - threshold._1, elapsedTime)); + inlineCompactDeltaSecondsMax)); } - return compactable; + break; default: - throw new HoodieCompactionException("Unsupported compact type: " + config.getInlineCompactTriggerStrategy()); + throw new HoodieCompactionException("Unsupported compaction trigger strategy: " + config.getInlineCompactTriggerStrategy()); } + return compactable; } public Long parsedToSeconds(String time) { diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index 1e7b37272a2cc..9e99e26c1814e 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -43,7 +43,7 @@ private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(true) .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommits) - .withMaxDeltaTimeBeforeCompaction(maxDeltaTime) + .withMaxDeltaSecondsBeforeCompaction(maxDeltaTime) .withInlineCompactionTriggerStrategy(inlineCompactionType).build()) .build(); } @@ -99,8 +99,7 @@ public void testSuccessfulCompactionBasedOnTime() throws Exception { runNextDeltaCommits(writeClient, readClient, Arrays.asList(instantTime), records, cfg, true, new ArrayList<>()); // after 10s, that will trigger compaction - Thread.sleep(10000); - String finalInstant = HoodieActiveTimeline.createNewInstantTime(); + String finalInstant = HoodieActiveTimeline.createNewInstantTime(10000); HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 100), writeClient, metaClient, cfg, false); @@ -127,10 +126,9 @@ public void testSuccessfulCompactionBasedOnNumOrTime() throws Exception { metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); assertEquals(4, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); - Thread.sleep(20000); // 4th commit, that will trigger compaction because reach the time elapsed metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); - finalInstant = HoodieActiveTimeline.createNewInstantTime(); + finalInstant = HoodieActiveTimeline.createNewInstantTime(20000); createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 10), writeClient, metaClient, cfg, false); metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); @@ -151,10 +149,9 @@ public void testSuccessfulCompactionBasedOnNumAndTime() throws Exception { // Then: ensure no compaction is executedm since there are only 3 delta commits assertEquals(3, metaClient.getActiveTimeline().getCommitsAndCompactionTimeline().countInstants()); - Thread.sleep(20000); // 4th commit, that will trigger compaction metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); - String finalInstant = HoodieActiveTimeline.createNewInstantTime(); + String finalInstant = HoodieActiveTimeline.createNewInstantTime(20000); createNextDeltaCommit(finalInstant, dataGen.generateUpdates(finalInstant, 10), writeClient, metaClient, cfg, false); metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); @@ -202,7 +199,7 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { HoodieWriteConfig cfg = getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(false) - .withMaxDeltaTimeBeforeCompaction(5) + .withMaxDeltaSecondsBeforeCompaction(5) .withInlineCompactionTriggerStrategy(CompactionTriggerStrategy.TIME_ELAPSED).build()) .build(); String instantTime; @@ -212,8 +209,7 @@ public void testCompactionRetryOnFailureBasedOnTime() throws Exception { HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); runNextDeltaCommits(writeClient, readClient, instants, records, cfg, true, new ArrayList<>()); // Schedule compaction instantTime, make it in-flight (simulates inline compaction failing) - Thread.sleep(10000); - instantTime = HoodieActiveTimeline.createNewInstantTime(); + instantTime = HoodieActiveTimeline.createNewInstantTime(10000); scheduleCompaction(instantTime, writeClient, cfg); moveCompactionFromRequestedToInflight(instantTime, cfg); } @@ -239,7 +235,7 @@ public void testCompactionRetryOnFailureBasedOnNumAndTime() throws Exception { HoodieWriteConfig cfg = getConfigBuilder(false) .withCompactionConfig(HoodieCompactionConfig.newBuilder() .withInlineCompaction(false) - .withMaxDeltaTimeBeforeCompaction(1) + .withMaxDeltaSecondsBeforeCompaction(1) .withMaxNumDeltaCommitsBeforeCompaction(1) .withInlineCompactionTriggerStrategy(CompactionTriggerStrategy.NUM_AND_TIME).build()) .build(); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java index fcb4fd9176dda..c8a49fdc4422a 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java @@ -90,6 +90,20 @@ public static String createNewInstantTime() { }); } + /** + * Returns next instant time that adds milliseconds in the {@link #COMMIT_FORMATTER} format. + * Ensures each instant time is atleast 1 second apart since we create instant times at second granularity + */ + public static String createNewInstantTime(long milliseconds) { + return lastInstantTime.updateAndGet((oldVal) -> { + String newCommitTime; + do { + newCommitTime = HoodieActiveTimeline.COMMIT_FORMATTER.format(new Date(System.currentTimeMillis() + milliseconds)); + } while (HoodieTimeline.compareTimestamps(newCommitTime, LESSER_THAN_OR_EQUALS, oldVal)); + return newCommitTime; + }); + } + protected HoodieActiveTimeline(HoodieTableMetaClient metaClient, Set includedExtensions) { this(metaClient, includedExtensions, true); } From 1ffe0f6b0f59991fcec6e8d99ca98da4d62760c5 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Fri, 5 Feb 2021 15:40:40 +0800 Subject: [PATCH 13/14] update --- .../hudi/config/HoodieCompactionConfig.java | 3 +- .../compact/CompactionTriggerStrategy.java | 9 ++++- ...SparkScheduleCompactionActionExecutor.java | 35 ++++++++----------- .../action/compact/TestInlineCompaction.java | 6 ++-- .../table/timeline/HoodieActiveTimeline.java | 10 ++---- 5 files changed, 30 insertions(+), 33 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 00b8d5eec73d1..934d91a274c26 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -47,6 +47,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { public static final String INLINE_COMPACT_PROP = "hoodie.compact.inline"; // Run a compaction every N delta commits public static final String INLINE_COMPACT_NUM_DELTA_COMMITS_PROP = "hoodie.compact.inline.max.delta.commits"; + // Run a compaction when time elapsed > N seconds since last compaction public static final String INLINE_COMPACT_TIME_DELTA_SECONDS_PROP = "hoodie.compact.inline.max.delta.seconds"; public static final String INLINE_COMPACT_TRIGGER_STRATEGY_PROP = "hoodie.compact.inline.trigger.strategy"; public static final String CLEANER_FILE_VERSIONS_RETAINED_PROP = "hoodie.cleaner.fileversions.retained"; @@ -113,7 +114,7 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig { private static final String DEFAULT_INCREMENTAL_CLEANER = "true"; private static final String DEFAULT_INLINE_COMPACT_NUM_DELTA_COMMITS = "5"; private static final String DEFAULT_INLINE_COMPACT_TIME_DELTA_SECONDS = String.valueOf(60 * 60); - private static final String DEFAULT_INLINE_COMPACT_TRIGGER_STRATEGY = CompactionTriggerStrategy.NUM.name(); + private static final String DEFAULT_INLINE_COMPACT_TRIGGER_STRATEGY = CompactionTriggerStrategy.NUM_COMMITS.name(); private static final String DEFAULT_CLEANER_FILE_VERSIONS_RETAINED = "3"; private static final String DEFAULT_CLEANER_COMMITS_RETAINED = "10"; private static final String DEFAULT_MAX_COMMITS_TO_KEEP = "30"; diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java index faba023b01472..6a4e634bc73a7 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/CompactionTriggerStrategy.java @@ -19,5 +19,12 @@ package org.apache.hudi.table.action.compact; public enum CompactionTriggerStrategy { - NUM, TIME_ELAPSED, NUM_AND_TIME, NUM_OR_TIME + // trigger compaction when reach N delta commits + NUM_COMMITS, + // trigger compaction when time elapsed > N seconds since last compaction + TIME_ELAPSED, + // trigger compaction when both NUM_COMMITS and TIME_ELAPSED are satisfied + NUM_AND_TIME, + // trigger compaction when NUM_COMMITS or TIME_ELAPSED is satisfied + NUM_OR_TIME } diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 1ec867cb39e58..29e408c24445c 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -83,58 +83,53 @@ protected HoodieCompactionPlan scheduleCompaction() { return new HoodieCompactionPlan(); } - public Tuple2 getLastDeltaCommitInfo(CompactionTriggerStrategy compactionTriggerStrategy) { + public Pair getLatestDeltaCommitInfo(CompactionTriggerStrategy compactionTriggerStrategy) { Option lastCompaction = table.getActiveTimeline().getCommitTimeline() .filterCompletedInstants().lastInstant(); HoodieTimeline deltaCommits = table.getActiveTimeline().getDeltaCommitTimeline(); - String lastCompactionTs; + String latestInstantTs; int deltaCommitsSinceLastCompaction = 0; if (lastCompaction.isPresent()) { - lastCompactionTs = lastCompaction.get().getTimestamp(); + latestInstantTs = lastCompaction.get().getTimestamp(); + deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfter(latestInstantTs, Integer.MAX_VALUE).countInstants(); } else { - lastCompactionTs = deltaCommits.firstInstant().get().getTimestamp(); + latestInstantTs = deltaCommits.firstInstant().get().getTimestamp(); + deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfterOrEquals(latestInstantTs, Integer.MAX_VALUE).countInstants(); } - if (compactionTriggerStrategy != CompactionTriggerStrategy.TIME_ELAPSED) { - if (lastCompaction.isPresent()) { - deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfter(lastCompactionTs, Integer.MAX_VALUE).countInstants(); - } else { - deltaCommitsSinceLastCompaction = deltaCommits.findInstantsAfterOrEquals(lastCompactionTs, Integer.MAX_VALUE).countInstants(); - } - } - return new Tuple2(deltaCommitsSinceLastCompaction, lastCompactionTs); + return Pair.of(deltaCommitsSinceLastCompaction, latestInstantTs); } public boolean needCompact(CompactionTriggerStrategy compactionTriggerStrategy) { boolean compactable; // get deltaCommitsSinceLastCompaction and lastCompactionTs - Tuple2 lastDeltaCommitInfo = getLastDeltaCommitInfo(compactionTriggerStrategy); + Pair latestDeltaCommitInfo = getLatestDeltaCommitInfo(compactionTriggerStrategy); int inlineCompactDeltaCommitMax = config.getInlineCompactDeltaCommitMax(); int inlineCompactDeltaSecondsMax = config.getInlineCompactDeltaSecondsMax(); switch (compactionTriggerStrategy) { - case NUM: - compactable = inlineCompactDeltaCommitMax <= lastDeltaCommitInfo._1; + case NUM_COMMITS: + compactable = inlineCompactDeltaCommitMax <= latestDeltaCommitInfo.getLeft(); if (compactable) { LOG.info(String.format("The delta commits >= %s, trigger compaction scheduler.", inlineCompactDeltaCommitMax)); } break; case TIME_ELAPSED: - compactable = inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(lastDeltaCommitInfo._2); + compactable = inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(latestDeltaCommitInfo.getRight()); if (compactable) { LOG.info(String.format("The elapsed time >=%ss, trigger compaction scheduler.", inlineCompactDeltaSecondsMax)); } break; case NUM_OR_TIME: - compactable = inlineCompactDeltaCommitMax <= lastDeltaCommitInfo._1 - || inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(lastDeltaCommitInfo._2); + compactable = inlineCompactDeltaCommitMax <= latestDeltaCommitInfo.getLeft() + || inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(latestDeltaCommitInfo.getRight()); if (compactable) { LOG.info(String.format("The delta commits >= %s or elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, inlineCompactDeltaSecondsMax)); } break; case NUM_AND_TIME: - compactable = inlineCompactDeltaCommitMax <= lastDeltaCommitInfo._1 - && inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(lastDeltaCommitInfo._2); + compactable = inlineCompactDeltaCommitMax <= latestDeltaCommitInfo.getLeft() + && inlineCompactDeltaSecondsMax <= parsedToSeconds(instantTime) - parsedToSeconds(latestDeltaCommitInfo.getRight()); if (compactable) { LOG.info(String.format("The delta commits >= %s and elapsed_time >=%ss, trigger compaction scheduler.", inlineCompactDeltaCommitMax, inlineCompactDeltaSecondsMax)); diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java index 9e99e26c1814e..80542edfa7f15 100644 --- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java +++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/action/compact/TestInlineCompaction.java @@ -51,7 +51,7 @@ private HoodieWriteConfig getConfigForInlineCompaction(int maxDeltaCommits, int @Test public void testCompactionIsNotScheduledEarly() throws Exception { // Given: make two commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.NUM); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.NUM_COMMITS); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { List records = dataGen.generateInserts(HoodieActiveTimeline.createNewInstantTime(), 100); HoodieReadClient readClient = getHoodieReadClient(cfg.getBasePath()); @@ -67,7 +67,7 @@ public void testCompactionIsNotScheduledEarly() throws Exception { @Test public void testSuccessfulCompactionBasedOnNumCommits() throws Exception { // Given: make three commits - HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.NUM); + HoodieWriteConfig cfg = getConfigForInlineCompaction(3, 60, CompactionTriggerStrategy.NUM_COMMITS); List instants = IntStream.range(0, 2).mapToObj(i -> HoodieActiveTimeline.createNewInstantTime()).collect(Collectors.toList()); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(cfg)) { @@ -180,7 +180,7 @@ public void testCompactionRetryOnFailureBasedOnNumCommits() throws Exception { } // When: a third commit happens - HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactionTriggerStrategy.NUM); + HoodieWriteConfig inlineCfg = getConfigForInlineCompaction(2, 60, CompactionTriggerStrategy.NUM_COMMITS); String instantTime3 = HoodieActiveTimeline.createNewInstantTime(); try (SparkRDDWriteClient writeClient = getHoodieWriteClient(inlineCfg)) { HoodieTableMetaClient metaClient = new HoodieTableMetaClient(hadoopConf, cfg.getBasePath()); diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java index c8a49fdc4422a..846eabe566a01 100644 --- a/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java +++ b/hudi-common/src/main/java/org/apache/hudi/common/table/timeline/HoodieActiveTimeline.java @@ -81,17 +81,11 @@ public class HoodieActiveTimeline extends HoodieDefaultTimeline { * Ensures each instant time is atleast 1 second apart since we create instant times at second granularity */ public static String createNewInstantTime() { - return lastInstantTime.updateAndGet((oldVal) -> { - String newCommitTime; - do { - newCommitTime = HoodieActiveTimeline.COMMIT_FORMATTER.format(new Date()); - } while (HoodieTimeline.compareTimestamps(newCommitTime, LESSER_THAN_OR_EQUALS, oldVal)); - return newCommitTime; - }); + return createNewInstantTime(0); } /** - * Returns next instant time that adds milliseconds in the {@link #COMMIT_FORMATTER} format. + * Returns next instant time that adds N milliseconds in the {@link #COMMIT_FORMATTER} format. * Ensures each instant time is atleast 1 second apart since we create instant times at second granularity */ public static String createNewInstantTime(long milliseconds) { From b34c416c83f5d651cbf9d10fe35c0047d69fc964 Mon Sep 17 00:00:00 2001 From: Karl-WangSK Date: Fri, 5 Feb 2021 16:28:56 +0800 Subject: [PATCH 14/14] update --- .../action/compact/SparkScheduleCompactionActionExecutor.java | 1 - 1 file changed, 1 deletion(-) diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java index 29e408c24445c..bf878b7a210f0 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/compact/SparkScheduleCompactionActionExecutor.java @@ -37,7 +37,6 @@ import org.apache.log4j.LogManager; import org.apache.log4j.Logger; import org.apache.spark.api.java.JavaRDD; -import scala.Tuple2; import java.io.IOException; import java.text.ParseException;