feat(clean): Adding empty clean support to hudi

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCleanConfig.java

@@ -250,6 +250,16 @@ public class HoodieCleanConfig extends HoodieConfig {
       .markAdvanced()
       .withDocumentation("Maximum number of commits to clean in one clean commit. Applicable only when the clean policy is based on KEEP_LATEST_COMMITS or KEEP_LATEST_HOURS");
 
+  public static final ConfigProperty<Long> MAX_INTERVAL_TO_CREATE_EMPTY_CLEAN_HOURS = ConfigProperty
+      .key("hoodie.write.empty.clean.internval.hours")
+      .defaultValue(-1L)
+      .markAdvanced()
+      .withDocumentation("In some cases empty clean commit needs to be created to ensure the clean planner "
+          + "does not look through entire dataset if there are no clean plans. This is possible for append-only "
+          + "dataset. Also, for these datasets we cannot ignore clean completely since in the future there could "
+          + "be upsert or replace operations. By creating empty clean commit, earliest_commit_to_retain value "
+          + "will be updated so that now clean planner can only check for partitions that are modified after the "
+          + "last empty clean's earliest_commit_toRetain value thereby optimizing the clean planning");
 
   /** @deprecated Use {@link #CLEANER_POLICY} and its methods instead */
   @Deprecated
@@ -426,6 +436,11 @@ public HoodieCleanConfig.Builder withMaxCommitsToClean(long maxCommitsToClean) {
       return this;
     }
 
+    public HoodieCleanConfig.Builder withMaxIntervalToCreateEmptyCleanHours(long durationHours) {
+      cleanConfig.setValue(MAX_INTERVAL_TO_CREATE_EMPTY_CLEAN_HOURS, String.valueOf(durationHours));
+      return this;
+    }
+
     public HoodieCleanConfig build() {
       cleanConfig.setDefaults(HoodieCleanConfig.class.getName());
       HoodieCleaningPolicy.valueOf(cleanConfig.getString(CLEANER_POLICY));
@@ -434,6 +449,10 @@ public HoodieCleanConfig build() {
       if (maxCommitsToClean < 1) {
         throw new IllegalArgumentException(MAX_COMMITS_TO_CLEAN.key() + " must be >= 1, but was " + maxCommitsToClean);
       }
+      long maxDurationHours = cleanConfig.getLong(MAX_INTERVAL_TO_CREATE_EMPTY_CLEAN_HOURS);
+      if (maxDurationHours == 0 || maxDurationHours < -1) {
+        throw new IllegalArgumentException(MAX_INTERVAL_TO_CREATE_EMPTY_CLEAN_HOURS.key() + " must be >= 1, but was " + maxDurationHours);
+      }
       return cleanConfig;
     }
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -1842,6 +1842,10 @@ public boolean isAutoClean() {
     return getBoolean(HoodieCleanConfig.AUTO_CLEAN);
   }
 
+  public long maxIntervalToCreateEmptyCleanHours() {
+    return getLong(HoodieCleanConfig.MAX_INTERVAL_TO_CREATE_EMPTY_CLEAN_HOURS);
+  }
+
   public boolean shouldArchiveBeyondSavepoint() {
     return getBooleanOrDefault(HoodieArchivalConfig.ARCHIVE_BEYOND_SAVEPOINT);
   }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanActionExecutor.java

@@ -45,13 +45,15 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.hudi.common.util.CleanerUtils.CLEAN_METADATA_VERSION_2;
 import static org.apache.hudi.common.util.StringUtils.isNullOrEmpty;
 
 @Slf4j
@@ -134,9 +136,9 @@ private static Stream<Pair<String, PartitionCleanStat>> deleteFilesFunc(Iterator
    * @throws IllegalArgumentException if unknown cleaning policy is provided
    */
   List<HoodieCleanStat> clean(HoodieEngineContext context, HoodieCleanerPlan cleanerPlan) {
-    int cleanerParallelism = Math.min(
+    int cleanerParallelism = Math.max(1, Math.min(
         cleanerPlan.getFilePathsToBeDeletedPerPartition().values().stream().mapToInt(List::size).sum(),
-        config.getCleanerParallelism());
+        config.getCleanerParallelism()));
     log.info("Using cleanerParallelism: {}", cleanerParallelism);
 
     context.setJobStatus(this.getClass().getSimpleName(), "Perform cleaning of table: " + config.getTableName());
@@ -155,7 +157,7 @@ List<HoodieCleanStat> clean(HoodieEngineContext context, HoodieCleanerPlan clean
 
     List<String> partitionsToBeDeleted = table.getMetaClient().getTableConfig().isTablePartitioned() && cleanerPlan.getPartitionsToBeDeleted() != null
         ? cleanerPlan.getPartitionsToBeDeleted()
-        : new ArrayList<>();
+        : Collections.emptyList();
     partitionsToBeDeleted.forEach(entry -> {
       if (!isNullOrEmpty(entry)) {
         deleteFileAndGetResult(table.getStorage(), table.getMetaClient().getBasePath() + "/" + entry);
@@ -213,17 +215,18 @@ private HoodieCleanMetadata runClean(HoodieTable<T, I, K, O> table, HoodieInstan
       }
 
       List<HoodieCleanStat> cleanStats = clean(context, cleanerPlan);
+      table.getMetaClient().reloadActiveTimeline();
+      HoodieCleanMetadata metadata;
       if (cleanStats.isEmpty()) {
-        return HoodieCleanMetadata.newBuilder().build();
+        metadata = createEmptyCleanMetadata(cleanerPlan, inflightInstant, timer.endTimer());
+      } else {
+        metadata = CleanerUtils.convertCleanMetadata(
+            inflightInstant.requestedTime(),
+            Option.of(timer.endTimer()),
+            cleanStats,
+            cleanerPlan.getExtraMetadata()
+        );
       }
-
-      table.getMetaClient().reloadActiveTimeline();
-      HoodieCleanMetadata metadata = CleanerUtils.convertCleanMetadata(
-          inflightInstant.requestedTime(),
-          Option.of(timer.endTimer()),
-          cleanStats,
-          cleanerPlan.getExtraMetadata()
-      );
       this.txnManager.beginStateChange(Option.of(inflightInstant), Option.empty());
       writeTableMetadata(metadata, inflightInstant.requestedTime());
       table.getActiveTimeline().transitionCleanInflightToComplete(
@@ -238,6 +241,23 @@ private HoodieCleanMetadata runClean(HoodieTable<T, I, K, O> table, HoodieInstan
     }
   }
 
+  private static HoodieCleanMetadata createEmptyCleanMetadata(HoodieCleanerPlan cleanerPlan, HoodieInstant inflightInstant, long timeTakenMillis) {
+    ValidationUtils.checkArgument(cleanerPlan.getEarliestInstantToRetain() != null, "For empty cleans, earliest instant to retain can never be null");
+    HoodieCleanMetadata.Builder cleanMetadataBuilder = HoodieCleanMetadata.newBuilder()
+        .setStartCleanTime(inflightInstant.requestedTime())
+        .setTimeTakenInMillis(timeTakenMillis)
+        .setTotalFilesDeleted(0)
+        .setLastCompletedCommitTimestamp(cleanerPlan.getLastCompletedCommitTimestamp())
+        .setVersion(CLEAN_METADATA_VERSION_2)
+        .setPartitionMetadata(Collections.emptyMap())
+        .setExtraMetadata(cleanerPlan.getExtraMetadata())
+        .setBootstrapPartitionMetadata(Collections.emptyMap());
+    if (cleanerPlan.getEarliestInstantToRetain() != null) {
+      cleanMetadataBuilder.setEarliestCommitToRetain(cleanerPlan.getEarliestInstantToRetain().getTimestamp());
+    }
+    return cleanMetadataBuilder.build();
+  }
+
   @Override
   public HoodieCleanMetadata execute() {
     List<HoodieCleanMetadata> cleanMetadataList = new ArrayList<>();

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/clean/CleanPlanActionExecutor.java

@@ -25,12 +25,13 @@
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.engine.HoodieLocalEngineContext;
 import org.apache.hudi.common.model.CleanFileInfo;
-import org.apache.hudi.common.model.HoodieCleaningPolicy;
 import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.table.timeline.HoodieInstantTimeGenerator;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.CleanerUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
@@ -42,13 +43,18 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
+import java.text.ParseException;
+import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import static org.apache.hudi.common.table.timeline.InstantComparison.LESSER_THAN;
+import static org.apache.hudi.common.table.timeline.InstantComparison.compareTimestamps;
 import static org.apache.hudi.common.util.CleanerUtils.SAVEPOINTED_TIMESTAMPS;
 import static org.apache.hudi.common.util.MapUtils.nonEmpty;
 
@@ -94,6 +100,23 @@ private boolean needsCleaning(CleaningTriggerStrategy strategy) {
     }
   }
 
+  private HoodieCleanerPlan getEmptyCleanerPlan(Option<HoodieInstant> earliestInstant, CleanPlanner<T, I, K, O> planner) throws IOException {
+    HoodieCleanerPlan.Builder cleanBuilder = HoodieCleanerPlan.newBuilder()
+        .setFilePathsToBeDeletedPerPartition(Collections.emptyMap())
+        .setExtraMetadata(prepareExtraMetadata(planner.getSavepointedTimestamps()));
+    if (earliestInstant.isPresent()) {
+      HoodieInstant hoodieInstant = earliestInstant.get();
+      cleanBuilder.setPolicy(config.getCleanerPolicy().name())
+          .setVersion(CleanPlanner.LATEST_CLEAN_PLAN_VERSION)
+          .setEarliestInstantToRetain(new HoodieActionInstant(hoodieInstant.requestedTime(), hoodieInstant.getAction(), hoodieInstant.getState().name()))
+          .setLastCompletedCommitTimestamp(planner.getLastCompletedCommitTimestamp());
+    } else {
+      cleanBuilder.setPolicy(config.getCleanerPolicy().name())
+          .setVersion(CleanPlanner.LATEST_CLEAN_PLAN_VERSION);
+    }
+    return cleanBuilder.build();
+  }
+
   /**
    * Generates List of files to be cleaned.
    *
@@ -109,8 +132,8 @@ HoodieCleanerPlan requestClean(HoodieEngineContext context) {
       context.clearJobStatus();
 
       if (partitionsToClean.isEmpty()) {
-        log.info("Nothing to clean here. It is already clean");
-        return HoodieCleanerPlan.newBuilder().setPolicy(HoodieCleaningPolicy.KEEP_LATEST_COMMITS.name()).build();
+        log.info("Partitions to clean returned empty. Checking to see if empty clean needs to be created.");
+        return getEmptyCleanerPlan(earliestInstant, planner);
       }
       log.info(
           "Earliest commit to retain for clean : {}",
@@ -213,14 +236,61 @@ protected Option<HoodieCleanerPlan> requestClean() {
       cleanerEngineContext = context;
     }
     final HoodieCleanerPlan cleanerPlan = requestClean(cleanerEngineContext);
-    Option<HoodieCleanerPlan> option = Option.empty();
-    if (nonEmpty(cleanerPlan.getFilePathsToBeDeletedPerPartition())
-        && cleanerPlan.getFilePathsToBeDeletedPerPartition().values().stream().mapToInt(List::size).sum() > 0) {
+    Option<HoodieCleanerPlan> cleanPlanOpt = Option.empty();
+    if ((cleanerPlan.getPartitionsToBeDeleted() != null && !cleanerPlan.getPartitionsToBeDeleted().isEmpty())
+        || (nonEmpty(cleanerPlan.getFilePathsToBeDeletedPerPartition())
+        && cleanerPlan.getFilePathsToBeDeletedPerPartition().values().stream().mapToInt(List::size).sum() > 0)) {
       // Only create cleaner plan which does some work
-      option = Option.of(cleanerPlan);
+      cleanPlanOpt = Option.of(cleanerPlan);
     }
+    // If cleaner plan returned an empty list, incremental clean is enabled and there was no
+    // completed clean created in the last X hours configured in MAX_DURATION_TO_CREATE_EMPTY_CLEAN,
+    // create a dummy clean to avoid full scan in the future.
+    // Note: For a dataset with incremental clean enabled, that does not receive any updates, cleaner plan always comes
+    // with an empty list of files to be cleaned.  CleanActionExecutor would never be invoked for this dataset.
+    // To avoid fullscan on the dataset with every ingestion run, empty clean commit is created here.
+    if (cleanPlanOpt.isEmpty() && config.incrementalCleanerModeEnabled() && cleanerPlan.getEarliestInstantToRetain() != null && config.maxIntervalToCreateEmptyCleanHours() > 0) {
+      // Only create an empty clean commit if earliestInstantToRetain is present in the plan
+      boolean eligibleForEmptyCleanCommit = true;
+
+      // if there is no previous clean instant or the previous clean instant was before the configured max duration, schedule an empty clean commit
+      Option<HoodieInstant> lastCleanInstant = table.getCleanTimeline().filterCompletedInstants().lastInstant();
+      if (lastCleanInstant.isPresent()) {
+        try {
+          ZonedDateTime latestDateTime = ZonedDateTime.ofInstant(java.time.Instant.now(), table.getMetaClient().getTableConfig().getTimelineTimezone().getZoneId());
+          long currentCleanTimeMs = latestDateTime.toInstant().toEpochMilli();
+          long lastCleanTimeMs = HoodieInstantTimeGenerator.parseDateFromInstantTime(lastCleanInstant.get().requestedTime()).toInstant().toEpochMilli();
+          eligibleForEmptyCleanCommit = currentCleanTimeMs - lastCleanTimeMs > (TimeUnit.HOURS.toMillis(config.maxIntervalToCreateEmptyCleanHours()));
+        } catch (ParseException e) {
+          log.error("Unable to parse last clean commit time", e);
+          throw new HoodieException("Unable to parse last clean commit time", e);
+        }
+      }
+      if (eligibleForEmptyCleanCommit) {
+        // Ensure earliestCommitToRetain doesn't go backwards when user changes cleaner configuration
+        if (lastCleanInstant.isPresent()) {
+          try {
+            HoodieCleanMetadata lastCleanMetadata = table.getActiveTimeline().readCleanMetadata(lastCleanInstant.get());
+            String previousEarliestCommitToRetain = lastCleanMetadata.getEarliestCommitToRetain();
+            String currentEarliestCommitToRetain = cleanerPlan.getEarliestInstantToRetain().getTimestamp();
 
-    return option;
+            if (!StringUtils.isNullOrEmpty(previousEarliestCommitToRetain) && !StringUtils.isNullOrEmpty(currentEarliestCommitToRetain)
+                && compareTimestamps(currentEarliestCommitToRetain, LESSER_THAN, previousEarliestCommitToRetain)) {
+              log.warn("Skipping empty clean creation because earliestCommitToRetain would go backwards. "
+                  + "Previous: {}, Current: {}. This can happen when cleaner configuration is changed.",
+                  previousEarliestCommitToRetain, currentEarliestCommitToRetain);
+              return Option.empty();
+            }
+          } catch (IOException e) {
+            log.error("Unable to read last clean metadata", e);
+            throw new HoodieException("Unable to read last clean metadata", e);
+          }
+        }
+        log.info("Creating an empty clean instant with earliestCommitToRetain of {}", cleanerPlan.getEarliestInstantToRetain().getTimestamp());
+        return Option.of(cleanerPlan);
+      }
+    }
+    return cleanPlanOpt;
   }
 
   @Override

hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/action/TestCleanPlanner.java

@@ -513,6 +513,14 @@ static Stream<Arguments> keepLatestByHoursOrCommitsArgsIncrCleanPartitions() {
         Collections.singletonMap(savepoint2, Collections.singletonList(PARTITION1)), Option.empty(),
         activeInstantsPartitionsMap2, Collections.emptyList(), threePartitionsInActiveTimeline, true, Collections.emptyMap()));
 
+    // Empty cleaner plan case
+    arguments.add(Arguments.of(true, getCleanByHoursConfig(), earliestInstant, lastCompletedInLastClean, lastCleanInstant,
+        earliestInstantInLastClean, Collections.emptyList(), Collections.emptyMap(), Option.empty(),
+        activeInstantsPartitionsMap2, Collections.emptyList(), twoPartitionsInActiveTimeline, false, Collections.emptyMap()));
+    arguments.add(Arguments.of(false, getCleanByHoursConfig(), earliestInstant, lastCompletedInLastClean, lastCleanInstant,
+        earliestInstantInLastClean, Collections.emptyList(), Collections.emptyMap(), Option.empty(),
+        activeInstantsUnPartitionsMap, Collections.emptyList(), unPartitionsInActiveTimeline, false, Collections.emptyMap()));
+
     return arguments.stream();
   }
 
@@ -598,8 +606,8 @@ private static HoodieCleanMetadata getCleanCommitMetadata(List<String> partition
         Collections.emptyList(), Collections.emptyList(), Collections.emptyList(), false)));
     Map<String, String> extraMetadata = new HashMap<>();
     extraMetadata.put(SAVEPOINTED_TIMESTAMPS, savepointsToTrack.stream().collect(Collectors.joining(",")));
-    return new HoodieCleanMetadata(instantTime, 100L, 10, earliestCommitToRetain, lastCompletedTime, partitionMetadata,
-        CLEAN_METADATA_VERSION_2, Collections.EMPTY_MAP, extraMetadata.isEmpty() ? null : extraMetadata);
+    return new HoodieCleanMetadata(instantTime, 100L, partitionMetadata.isEmpty() ? 0 : 10, earliestCommitToRetain, lastCompletedTime,
+        partitionMetadata, CLEAN_METADATA_VERSION_2, Collections.emptyMap(), extraMetadata.isEmpty() ? null : extraMetadata);
   }
 
   private static HoodieSavepointMetadata getSavepointMetadata(List<String> partitions) {