apache · vinothchandar · Jul 1, 2020
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/HoodieReadClient.java b/hudi-client/src/main/java/org/apache/hudi/client/HoodieReadClient.java
@@ -19,12 +19,14 @@
 package org.apache.hudi.client;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
 import org.apache.hudi.common.model.HoodieBaseFile;
 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.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ClusteringUtils;
 import org.apache.hudi.common.util.CompactionUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.collection.Pair;
@@ -197,10 +199,17 @@ public JavaRDD<HoodieRecord<T>> tagLocation(JavaRDD<HoodieRecord<T>> hoodieRecor
    */
   public List<Pair<String, HoodieCompactionPlan>> getPendingCompactions() {
     HoodieTableMetaClient metaClient =
-        new HoodieTableMetaClient(hadoopConf, hoodieTable.getMetaClient().getBasePath(), true);
+            new HoodieTableMetaClient(hadoopConf, hoodieTable.getMetaClient().getBasePath(), true);
     return CompactionUtils.getAllPendingCompactionPlans(metaClient).stream()
-        .map(
-            instantWorkloadPair -> Pair.of(instantWorkloadPair.getKey().getTimestamp(), instantWorkloadPair.getValue()))
-        .collect(Collectors.toList());
+            .map(instantWorkloadPair -> Pair.of(instantWorkloadPair.getKey().getTimestamp(), instantWorkloadPair.getValue()))
+            .collect(Collectors.toList());
+  }
+
+  public List<Pair<String, HoodieClusteringPlan>> getPendingClustering() {
+    HoodieTableMetaClient metaClient =
+            new HoodieTableMetaClient(hadoopConf, hoodieTable.getMetaClient().getBasePath(), true);
+    return ClusteringUtils.getAllPendingClusteringPlans(metaClient).stream()
+            .map(instantWorkloadPair -> Pair.of(instantWorkloadPair.getKey().getTimestamp(), instantWorkloadPair.getValue()))
+            .collect(Collectors.toList());
   }
 }
diff --git a/hudi-client/src/main/java/org/apache/hudi/client/HoodieWriteClient.java b/hudi-client/src/main/java/org/apache/hudi/client/HoodieWriteClient.java
@@ -19,6 +19,7 @@
 package org.apache.hudi.client;
 
 import org.apache.hudi.avro.model.HoodieCleanMetadata;
+import org.apache.hudi.avro.model.HoodieClusteringPlan;
 import org.apache.hudi.avro.model.HoodieCompactionPlan;
 import org.apache.hudi.avro.model.HoodieRestoreMetadata;
 import org.apache.hudi.avro.model.HoodieRollbackMetadata;
@@ -49,6 +50,7 @@
 import org.apache.hudi.table.HoodieTimelineArchiveLog;
 import org.apache.hudi.table.UserDefinedBulkInsertPartitioner;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
+import org.apache.hudi.table.action.clustering.ClusteringHelpers;
 import org.apache.hudi.table.action.compact.CompactHelpers;
 import org.apache.hudi.table.action.savepoint.SavepointHelpers;
 
@@ -80,6 +82,7 @@ public class HoodieWriteClient<T extends HoodieRecordPayload> extends AbstractHo
   private final boolean rollbackPending;
   private final transient HoodieMetrics metrics;
   private transient Timer.Context compactionTimer;
+  private transient Timer.Context clusteringTimer;
   private transient AsyncCleanerService asyncCleanerService;
 
   /**
@@ -342,6 +345,11 @@ protected void postCommit(HoodieCommitMetadata metadata, String instantTime,
       } else {
         metadata.addMetadata(HoodieCompactionConfig.INLINE_COMPACT_PROP, "false");
       }
+      if (config.isInlineClustering()) {
+        inlineClustering(extraMetadata);
+      } else {
+        metadata.addMetadata(HoodieCompactionConfig.INLINE_CLUSTERING_PROP, "false");
+      }
       // We cannot have unbounded commit files. Archive commits if we have to archive
       HoodieTimelineArchiveLog archiveLog = new HoodieTimelineArchiveLog(config, createMetaClient(true));
       archiveLog.archiveIfRequired(hadoopConf);
@@ -565,6 +573,18 @@ private void startCommit(String instantTime) {
         instantTime));
   }
 
+  public Option<String> scheduleClustering(Option<Map<String, String>> extraMetadata) throws HoodieIOException {
+    String instantTime = HoodieActiveTimeline.createNewInstantTime();
+    return scheduleClusteringAtInstant(instantTime, extraMetadata) ? Option.of(instantTime) : Option.empty();
+  }
+
+  public boolean scheduleClusteringAtInstant(String instantTime, Option<Map<String, String>> extraMetadata) throws HoodieIOException {
+    LOG.info("Scheduling clustering at instant time :" + instantTime);
+    Option<HoodieClusteringPlan> plan = HoodieTable.create(config, hadoopConf)
+            .scheduleClustering(jsc, instantTime, extraMetadata);
+    return plan.isPresent();
+  }
+
   /**
    * Schedules a new compaction instant.
    *
@@ -598,6 +618,10 @@ public JavaRDD<WriteStatus> compact(String compactionInstantTime) {
     return compact(compactionInstantTime, config.shouldAutoCommit());
   }
 
+  public JavaRDD<WriteStatus> clustering(String clusteringInstantTime) {
+    return clustering(clusteringInstantTime, config.shouldAutoCommit());
+  }
+
   /**
    * Commit a compaction operation. Allow passing additional meta-data to be stored in commit instant file.
    *
@@ -638,6 +662,27 @@ protected void completeCompaction(HoodieCommitMetadata metadata, JavaRDD<WriteSt
     LOG.info("Compacted successfully on commit " + compactionCommitTime);
   }
 
+  protected void completeClustering(HoodieCommitMetadata metadata, JavaRDD<WriteStatus> writeStatuses, HoodieTable<T> table,
+                                    String clusteringCommitTime) {
+
+    List<HoodieWriteStat> writeStats = writeStatuses.map(WriteStatus::getStat).collect();
+    finalizeWrite(table, clusteringCommitTime, writeStats);
+    LOG.info("Committing Clustering " + clusteringCommitTime + ". Finished with result " + metadata);
+    ClusteringHelpers.completeInflightClustering(table, clusteringCommitTime, metadata);
+
+    if (clusteringTimer != null) {
+      long durationInMs = metrics.getDurationInMs(clusteringTimer.stop());
+      try {
+        metrics.updateCommitMetrics(HoodieActiveTimeline.COMMIT_FORMATTER.parse(clusteringCommitTime).getTime(),
+                durationInMs, metadata, HoodieActiveTimeline.COMPACTION_ACTION);
+      } catch (ParseException e) {
+        throw new HoodieCommitException("Commit time is not of valid format. Failed to commit compaction "
+                + config.getBasePath() + " at time " + clusteringCommitTime, e);
+      }
+    }
+    LOG.info("Clustering successfully on commit " + clusteringCommitTime);
+  }
+
   /**
    * Rollback failed compactions. Inflight rollbacks for compactions revert the .inflight file to the .requested file
    *
@@ -649,6 +694,11 @@ public void rollbackInflightCompaction(HoodieInstant inflightInstant, HoodieTabl
     table.getActiveTimeline().revertCompactionInflightToRequested(inflightInstant);
   }
 
+  public void rollbackInflightClustering(HoodieInstant inflightInstant, HoodieTable table) {
+    table.rollback(jsc, HoodieActiveTimeline.createNewInstantTime(), inflightInstant, false);
+    table.getActiveTimeline().revertClusteringInflightToRequested(inflightInstant);
+  }
+
   /**
    * Cleanup all pending commits.
    */
@@ -685,6 +735,23 @@ private JavaRDD<WriteStatus> compact(String compactionInstantTime, boolean shoul
     return statuses;
   }
 
+  private JavaRDD<WriteStatus> clustering(String clusteringInstantTime, boolean shouldComplete) {
+    HoodieTable<T> table = HoodieTable.create(config, hadoopConf);
+    HoodieTimeline pendingClusteringTimeline = table.getActiveTimeline().filterPendingClusteringTimeline();
+    HoodieInstant inflightInstant = HoodieTimeline.getClusteringInflightInstant(clusteringInstantTime);
+    if (pendingClusteringTimeline.containsInstant(inflightInstant)) {
+      rollbackInflightClustering(inflightInstant, table);
+      table.getMetaClient().reloadActiveTimeline();
+    }
+    clusteringTimer = metrics.getClusteringCtx();
+    HoodieWriteMetadata clusteringMetadata = table.clustering(jsc, clusteringInstantTime);
+    JavaRDD<WriteStatus> statuses = clusteringMetadata.getWriteStatuses();
+    if (shouldComplete && clusteringMetadata.getCommitMetadata().isPresent()) {
+      completeClustering(clusteringMetadata.getCommitMetadata().get(), statuses, table, clusteringInstantTime);
+    }
+    return statuses;
+  }
+
   /**
    * Performs a compaction operation on a table, serially before or after an insert/upsert action.
    */
@@ -696,4 +763,12 @@ private Option<String> inlineCompact(Option<Map<String, String>> extraMetadata)
     });
     return compactionInstantTimeOpt;
   }
+
+  private Option<String> inlineClustering(Option<Map<String, String>> extraMetadata) {
+    Option<String> clusteringInstantTimeOpt = scheduleClustering(extraMetadata);
+    clusteringInstantTimeOpt.ifPresent(clusteringInstantTime -> {
+      clustering(clusteringInstantTime, true);
+    });
+    return clusteringInstantTimeOpt;
+  }
 }
diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java
@@ -22,6 +22,10 @@
 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.clustering.strategy.BaseFileSizeBasedClusteringStrategy;
+import org.apache.hudi.table.action.clustering.strategy.ClusteringStrategy;
+import org.apache.hudi.table.action.clustering.updates.RejectUpdateStrategy;
+import org.apache.hudi.table.action.clustering.updates.UpdateStrategy;
 import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
 import org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy;
 
@@ -44,8 +48,10 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig {
 
   // Turn on inline compaction - after fw delta commits a inline compaction will be run
   public static final String INLINE_COMPACT_PROP = "hoodie.compact.inline";
+  public static final String INLINE_CLUSTERING_PROP = "hoodie.clustering.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_CLUSTERING_NUM_DELTA_COMMITS_PROP = "hoodie.clustering.inline.max.delta.commits";
   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";
@@ -86,9 +92,17 @@ public class HoodieCompactionConfig extends DefaultHoodieConfig {
   public static final String TARGET_IO_PER_COMPACTION_IN_MB_PROP = "hoodie.compaction.target.io";
   // 500GB of target IO per compaction (both read and write)
   public static final String DEFAULT_TARGET_IO_PER_COMPACTION_IN_MB = String.valueOf(500 * 1024);
+  public static final String TARGET_IO_PER_CLUSTERING_IN_MB_PROP = "hoodie.clustering.target.io";
+  // 500GB of target IO per clustering (both read and write)
+  public static final String DEFAULT_TARGET_IO_PER_CLUSTERING_IN_MB = String.valueOf(500 * 1024 * 1024);
   public static final String COMPACTION_STRATEGY_PROP = "hoodie.compaction.strategy";
   // 200GB of target IO per compaction
   public static final String DEFAULT_COMPACTION_STRATEGY = LogFileSizeBasedCompactionStrategy.class.getName();
+  public static final String CLUSTERING_STRATEGY_PROP = "hoodie.clustering.strategy";
+  public static final String DEFAULT_CLUSTERING_STRATEGY = BaseFileSizeBasedClusteringStrategy.class.getName();
+  public static final String CLUSTERING_UPDATES_STRATEGY_PROP = "hoodie.clustering.updates.strategy";
+  public static final String DEFAULT_CLUSTERING_UPDATES_STRATEGY = RejectUpdateStrategy.class.getName();
+
   // used to merge records written to log file
   public static final String DEFAULT_PAYLOAD_CLASS = OverwriteWithLatestAvroPayload.class.getName();
   public static final String PAYLOAD_CLASS_PROP = "hoodie.compaction.payload.class";
@@ -107,6 +121,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_CLUSTERING_NUM_DELTA_COMMITS = "5";
   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";
@@ -161,6 +176,11 @@ public Builder withInlineCompaction(Boolean inlineCompaction) {
       return this;
     }
 
+    public Builder withInlineClustering(Boolean inlineClustering) {
+      props.setProperty(INLINE_CLUSTERING_PROP, String.valueOf(inlineClustering));
+      return this;
+    }
+
     public Builder withCleanerPolicy(HoodieCleaningPolicy policy) {
       props.setProperty(CLEANER_POLICY_PROP, policy.name());
       return this;
@@ -212,6 +232,16 @@ public Builder withCleanerParallelism(int cleanerParallelism) {
       return this;
     }
 
+    public Builder withClusteringStrategy(ClusteringStrategy clusteringStrategy) {
+      props.setProperty(CLUSTERING_STRATEGY_PROP, clusteringStrategy.getClass().getName());
+      return this;
+    }
+
+    public Builder withClusteringUpdatesStrategy(UpdateStrategy updatesStrategy) {
+      props.setProperty(CLUSTERING_UPDATES_STRATEGY_PROP, updatesStrategy.getClass().getName());
+      return this;
+    }
+
     public Builder withCompactionStrategy(CompactionStrategy compactionStrategy) {
       props.setProperty(COMPACTION_STRATEGY_PROP, compactionStrategy.getClass().getName());
       return this;
@@ -232,6 +262,11 @@ public Builder withMaxNumDeltaCommitsBeforeCompaction(int maxNumDeltaCommitsBefo
       return this;
     }
 
+    public Builder withMaxNumDeltaCommitsBeforeClustering(int maxNumDeltaCommitsBeforeClustering) {
+      props.setProperty(INLINE_CLUSTERING_NUM_DELTA_COMMITS_PROP, String.valueOf(maxNumDeltaCommitsBeforeClustering));
+      return this;
+    }
+
     public Builder withCompactionLazyBlockReadEnabled(Boolean compactionLazyBlockReadEnabled) {
       props.setProperty(COMPACTION_LAZY_BLOCK_READ_ENABLED_PROP, String.valueOf(compactionLazyBlockReadEnabled));
       return this;
@@ -263,6 +298,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_CLUSTERING_NUM_DELTA_COMMITS_PROP),
+          INLINE_CLUSTERING_NUM_DELTA_COMMITS_PROP, DEFAULT_INLINE_CLUSTERING_NUM_DELTA_COMMITS);
       setDefaultOnCondition(props, !props.containsKey(CLEANER_POLICY_PROP), CLEANER_POLICY_PROP,
           DEFAULT_CLEANER_POLICY);
       setDefaultOnCondition(props, !props.containsKey(CLEANER_FILE_VERSIONS_RETAINED_PROP),
@@ -285,11 +322,17 @@ public HoodieCompactionConfig build() {
           COPY_ON_WRITE_TABLE_RECORD_SIZE_ESTIMATE, DEFAULT_COPY_ON_WRITE_TABLE_RECORD_SIZE_ESTIMATE);
       setDefaultOnCondition(props, !props.containsKey(CLEANER_PARALLELISM), CLEANER_PARALLELISM,
           DEFAULT_CLEANER_PARALLELISM);
+      setDefaultOnCondition(props, !props.containsKey(CLUSTERING_STRATEGY_PROP), CLUSTERING_STRATEGY_PROP,
+          DEFAULT_CLUSTERING_STRATEGY);
+      setDefaultOnCondition(props, !props.containsKey(CLUSTERING_UPDATES_STRATEGY_PROP), CLUSTERING_UPDATES_STRATEGY_PROP,
+          DEFAULT_CLUSTERING_UPDATES_STRATEGY);
       setDefaultOnCondition(props, !props.containsKey(COMPACTION_STRATEGY_PROP), COMPACTION_STRATEGY_PROP,
           DEFAULT_COMPACTION_STRATEGY);
       setDefaultOnCondition(props, !props.containsKey(PAYLOAD_CLASS_PROP), PAYLOAD_CLASS_PROP, DEFAULT_PAYLOAD_CLASS);
       setDefaultOnCondition(props, !props.containsKey(TARGET_IO_PER_COMPACTION_IN_MB_PROP),
           TARGET_IO_PER_COMPACTION_IN_MB_PROP, DEFAULT_TARGET_IO_PER_COMPACTION_IN_MB);
+      setDefaultOnCondition(props, !props.containsKey(TARGET_IO_PER_CLUSTERING_IN_MB_PROP),
+          TARGET_IO_PER_CLUSTERING_IN_MB_PROP, DEFAULT_TARGET_IO_PER_CLUSTERING_IN_MB);
       setDefaultOnCondition(props, !props.containsKey(COMPACTION_LAZY_BLOCK_READ_ENABLED_PROP),
           COMPACTION_LAZY_BLOCK_READ_ENABLED_PROP, DEFAULT_COMPACTION_LAZY_BLOCK_READ_ENABLED);
       setDefaultOnCondition(props, !props.containsKey(COMPACTION_REVERSE_LOG_READ_ENABLED_PROP),

diff --git a/hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java b/hudi-client/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
@@ -29,6 +29,8 @@
 import org.apache.hudi.index.HoodieIndex;
 import org.apache.hudi.metrics.MetricsReporterType;
 import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite;
+import org.apache.hudi.table.action.clustering.strategy.ClusteringStrategy;
+import org.apache.hudi.table.action.clustering.updates.UpdateStrategy;
 import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
 
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
@@ -308,10 +310,26 @@ public boolean isInlineCompaction() {
     return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_PROP));
   }
 
+  public boolean isInlineClustering() {
+    return Boolean.parseBoolean(props.getProperty(HoodieCompactionConfig.INLINE_CLUSTERING_PROP));
+  }
+
   public int getInlineCompactDeltaCommitMax() {
     return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS_PROP));
   }
 
+  public int getInlineClusteringDeltaCommitMax() {
+    return Integer.parseInt(props.getProperty(HoodieCompactionConfig.INLINE_CLUSTERING_NUM_DELTA_COMMITS_PROP));
+  }
+
+  public ClusteringStrategy getClusteringStrategy() {
+    return ReflectionUtils.loadClass(props.getProperty(HoodieCompactionConfig.CLUSTERING_STRATEGY_PROP));
+  }
+
+  public UpdateStrategy getClusteringUpdatesStrategy() {
+    return ReflectionUtils.loadClass(props.getProperty(HoodieCompactionConfig.CLUSTERING_UPDATES_STRATEGY_PROP));
+  }
+
   public CompactionStrategy getCompactionStrategy() {
     return ReflectionUtils.loadClass(props.getProperty(HoodieCompactionConfig.COMPACTION_STRATEGY_PROP));
   }
@@ -320,6 +338,10 @@ public Long getTargetIOPerCompactionInMB() {
     return Long.parseLong(props.getProperty(HoodieCompactionConfig.TARGET_IO_PER_COMPACTION_IN_MB_PROP));
   }
 
+  public long getTargetIOPerClusteringInMB() {
+    return Long.parseLong(props.getProperty(HoodieCompactionConfig.TARGET_IO_PER_CLUSTERING_IN_MB_PROP));
+  }
+
   public Boolean getCompactionLazyBlockReadEnabled() {
     return Boolean.valueOf(props.getProperty(HoodieCompactionConfig.COMPACTION_LAZY_BLOCK_READ_ENABLED_PROP));
   }

diff --git a/hudi-client/src/main/java/org/apache/hudi/exception/HoodieClusteringException.java b/hudi-client/src/main/java/org/apache/hudi/exception/HoodieClusteringException.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 HoodieClusteringException extends HoodieException {
+
+  public HoodieClusteringException(String msg) {
+    super(msg);
+  }
+
+  public HoodieClusteringException(String msg, Throwable e) {
+    super(msg, e);
+  }
+}