a new way to config the merger

Author: gengxiaoyu

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

@@ -132,9 +132,15 @@ public class HoodieWriteConfig extends HoodieConfig {
   public static final ConfigProperty<String> MERGER_IMPLS = ConfigProperty
       .key("hoodie.datasource.write.merger.impls")
       .defaultValue(HoodieAvroRecordMerger.class.getName())
-      .withDocumentation("List of HoodieMerger implementations constituting Hudi's merging strategy -- based on the engine used "
+      .withDocumentation("List of HoodieMerger implementations constituting Hudi's merging strategy -- based on the engine used. "
+          + "These merger impls will filter by hoodie.datasource.write.merger.strategy "
           + "Hudi will pick most efficient implementation to perform merging/combining of the records (during update, reading MOR table, etc)");
 
+  public static final ConfigProperty<String> MERGER_STRATEGY = ConfigProperty
+      .key("hoodie.datasource.write.merger.strategy")
+      .defaultValue(StringUtils.DEFAULT_MERGER_STRATEGY_UUID)
+      .withDocumentation("Id of merger strategy.  Hudi will pick RecordMergers in hoodie.datasource.write.merger.impls which has the same merger strategy id");
+
   public static final ConfigProperty<String> KEYGENERATOR_CLASS_NAME = ConfigProperty
       .key("hoodie.datasource.write.keygenerator.class")
       .noDefaultValue()
@@ -912,7 +918,8 @@ private void applyMergerClass() {
         .map(String::trim)
         .distinct()
         .collect(Collectors.toList());
-    this.recordMerger = HoodieRecordUtils.generateRecordMerger(getString(BASE_PATH), engineType, mergers);
+    String mergerStrategy = getString(MERGER_STRATEGY);
+    this.recordMerger = HoodieRecordUtils.generateRecordMerger(getString(BASE_PATH), engineType, mergers, mergerStrategy);
   }
 
   public static HoodieWriteConfig.Builder newBuilder() {
@@ -939,7 +946,7 @@ public void setSchema(String schemaStr) {
   }
 
   public void setMergerClass(String mergerStrategy) {
-    setValue(MERGER_IMPLS, mergerStrategy);
+    setValue(MERGER_STRATEGY, mergerStrategy);
   }
 
   public String getInternalSchema() {
@@ -2258,6 +2265,11 @@ public Builder withMergerImpls(String mergerImpls) {
       return this;
     }
 
+    public Builder withMergerStrategy(String mergerStrategy) {
+      writeConfig.setValue(MERGER_STRATEGY, mergerStrategy);
+      return this;
+    }
+
     public Builder withKeyGenerator(String keyGeneratorClass) {
       writeConfig.setValue(KEYGENERATOR_CLASS_NAME, keyGeneratorClass);
       return this;

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecordMerger.java

@@ -23,6 +23,7 @@
 
 import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.metadata.HoodieMetadataPayload;
 
@@ -33,6 +34,11 @@
 
 public class HoodieAvroRecordMerger implements HoodieRecordMerger {
 
+  @Override
+  public String getMergingStrategy() {
+    return StringUtils.DEFAULT_MERGER_STRATEGY_UUID;
+  }
+
   @Override
   public Option<HoodieRecord> merge(HoodieRecord older, HoodieRecord newer, Schema schema, Properties props) throws IOException {
     ValidationUtils.checkArgument(older.getRecordType() == HoodieRecordType.AVRO);

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordMerger.java

@@ -49,4 +49,9 @@ public interface HoodieRecordMerger extends Serializable {
    * SPARK, AVRO, FLINK
    */
   HoodieRecordType getRecordType();
+
+  /**
+   * The kind of merging strategy this recordMerger belongs to. An UUID represents merging strategy.
+   */
+  String getMergingStrategy();
 }

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java

@@ -26,7 +26,6 @@
 import org.apache.hudi.common.config.HoodieConfig;
 import org.apache.hudi.common.config.OrderedProperties;
 import org.apache.hudi.common.config.TypedProperties;
-import org.apache.hudi.common.model.HoodieAvroRecordMerger;
 import org.apache.hudi.common.model.HoodieFileFormat;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieTableType;
@@ -156,11 +155,10 @@ public class HoodieTableConfig extends HoodieConfig {
       .withDocumentation("Payload class to use for performing compactions, i.e merge delta logs with current base file and then "
           + " produce a new base file.");
 
-  public static final ConfigProperty<String> MERGER_IMPLS = ConfigProperty
-      .key("hoodie.compaction.merger.impls")
-      .defaultValue(HoodieAvroRecordMerger.class.getName())
-      .withDocumentation("List of HoodieMerger implementations constituting Hudi's merging strategy -- based on the engine used "
-          + "Hudi will pick most efficient implementation to perform merging/combining of the records (during update, reading MOR table, etc)");
+  public static final ConfigProperty<String> MERGER_STRATEGY = ConfigProperty
+      .key("hoodie.compaction.merger.strategy")
+      .defaultValue(StringUtils.DEFAULT_MERGER_STRATEGY_UUID)
+      .withDocumentation("Id of merger strategy.  Hudi will pick RecordMergers in hoodie.datasource.write.merger.impls which has the same merger strategy id");
 
   public static final ConfigProperty<String> ARCHIVELOG_FOLDER = ConfigProperty
       .key("hoodie.archivelog.folder")
@@ -244,7 +242,7 @@ public class HoodieTableConfig extends HoodieConfig {
 
   private static final String TABLE_CHECKSUM_FORMAT = "%s.%s"; // <database_name>.<table_name>
 
-  public HoodieTableConfig(FileSystem fs, String metaPath, String payloadClassName, String mergerImpls) {
+  public HoodieTableConfig(FileSystem fs, String metaPath, String payloadClassName, String mergerStrategy) {
     super();
     Path propertyPath = new Path(metaPath, HOODIE_PROPERTIES_FILE);
     LOG.info("Loading table properties from " + propertyPath);
@@ -256,9 +254,9 @@ public HoodieTableConfig(FileSystem fs, String metaPath, String payloadClassName
         setValue(PAYLOAD_CLASS_NAME, payloadClassName);
         needStore = true;
       }
-      if (contains(MERGER_IMPLS) && payloadClassName != null
-          && !getString(MERGER_IMPLS).equals(mergerImpls)) {
-        setValue(MERGER_IMPLS, mergerImpls);
+      if (contains(MERGER_STRATEGY) && payloadClassName != null
+          && !getString(MERGER_STRATEGY).equals(mergerStrategy)) {
+        setValue(MERGER_STRATEGY, mergerStrategy);
         needStore = true;
       }
       if (needStore) {
@@ -428,7 +426,7 @@ public static void create(FileSystem fs, Path metadataFolder, Properties propert
       hoodieConfig.setDefaultValue(TYPE);
       if (hoodieConfig.getString(TYPE).equals(HoodieTableType.MERGE_ON_READ.name())) {
         hoodieConfig.setDefaultValue(PAYLOAD_CLASS_NAME);
-        hoodieConfig.setDefaultValue(MERGER_IMPLS);
+        hoodieConfig.setDefaultValue(MERGER_STRATEGY);
       }
       hoodieConfig.setDefaultValue(ARCHIVELOG_FOLDER);
       if (!hoodieConfig.contains(TIMELINE_LAYOUT_VERSION)) {
@@ -501,11 +499,8 @@ public String getPayloadClass() {
   /**
    * Read the payload class for HoodieRecords from the table properties.
    */
-  public String getMergerImpls() {
-    // There could be tables written with merge strategy from com.uber.hoodie. Need to transparently
-    // change to org.apache.hudi
-    return getStringOrDefault(MERGER_IMPLS).replace("com.uber.hoodie",
-        "org.apache.hudi");
+  public String getMergerStrategy() {
+    return getStringOrDefault(MERGER_STRATEGY);
   }
 
   public String getPreCombineField() {

hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java

@@ -116,7 +116,7 @@ public class HoodieTableMetaClient implements Serializable {
 
   protected HoodieTableMetaClient(Configuration conf, String basePath, boolean loadActiveTimelineOnLoad,
                                 ConsistencyGuardConfig consistencyGuardConfig, Option<TimelineLayoutVersion> layoutVersion,
-                                String payloadClassName, String mergerImpls, FileSystemRetryConfig fileSystemRetryConfig) {
+                                String payloadClassName, String mergerStrategy, FileSystemRetryConfig fileSystemRetryConfig) {
     LOG.info("Loading HoodieTableMetaClient from " + basePath);
     this.consistencyGuardConfig = consistencyGuardConfig;
     this.fileSystemRetryConfig = fileSystemRetryConfig;
@@ -125,7 +125,7 @@ protected HoodieTableMetaClient(Configuration conf, String basePath, boolean loa
     this.metaPath = new SerializablePath(new CachingPath(basePath, METAFOLDER_NAME));
     this.fs = getFs();
     TableNotFoundException.checkTableValidity(fs, this.basePath.get(), metaPath.get());
-    this.tableConfig = new HoodieTableConfig(fs, metaPath.toString(), payloadClassName, mergerImpls);
+    this.tableConfig = new HoodieTableConfig(fs, metaPath.toString(), payloadClassName, mergerStrategy);
     this.tableType = tableConfig.getTableType();
     Option<TimelineLayoutVersion> tableConfigVersion = tableConfig.getTimelineLayoutVersion();
     if (layoutVersion.isPresent() && tableConfigVersion.isPresent()) {
@@ -159,7 +159,7 @@ public static HoodieTableMetaClient reload(HoodieTableMetaClient oldMetaClient)
         .setConsistencyGuardConfig(oldMetaClient.consistencyGuardConfig)
         .setLayoutVersion(Option.of(oldMetaClient.timelineLayoutVersion))
         .setPayloadClassName(null)
-        .setMergerImpls(null)
+        .setMergerStrategy(null)
         .setFileSystemRetryConfig(oldMetaClient.fileSystemRetryConfig).build();
   }
 
@@ -633,7 +633,7 @@ public void initializeBootstrapDirsIfNotExists() throws IOException {
 
   private static HoodieTableMetaClient newMetaClient(Configuration conf, String basePath, boolean loadActiveTimelineOnLoad,
       ConsistencyGuardConfig consistencyGuardConfig, Option<TimelineLayoutVersion> layoutVersion,
-      String payloadClassName, String mergerImpls, FileSystemRetryConfig fileSystemRetryConfig, Properties props) {
+      String payloadClassName, String mergerStrategy, FileSystemRetryConfig fileSystemRetryConfig, Properties props) {
     HoodieMetastoreConfig metastoreConfig = null == props
         ? new HoodieMetastoreConfig.Builder().build()
         : new HoodieMetastoreConfig.Builder().fromProperties(props).build();
@@ -643,7 +643,7 @@ private static HoodieTableMetaClient newMetaClient(Configuration conf, String ba
             conf, consistencyGuardConfig, fileSystemRetryConfig,
             props.getProperty(HoodieTableConfig.DATABASE_NAME.key()), props.getProperty(HoodieTableConfig.NAME.key()), metastoreConfig)
         : new HoodieTableMetaClient(conf, basePath,
-        loadActiveTimelineOnLoad, consistencyGuardConfig, layoutVersion, payloadClassName, mergerImpls, fileSystemRetryConfig);
+        loadActiveTimelineOnLoad, consistencyGuardConfig, layoutVersion, payloadClassName, mergerStrategy, fileSystemRetryConfig);
   }
 
   public static Builder builder() {
@@ -659,7 +659,7 @@ public static class Builder {
     private String basePath;
     private boolean loadActiveTimelineOnLoad = false;
     private String payloadClassName = null;
-    private String mergerImpls = null;
+    private String mergerStrategy = null;
     private ConsistencyGuardConfig consistencyGuardConfig = ConsistencyGuardConfig.newBuilder().build();
     private FileSystemRetryConfig fileSystemRetryConfig = FileSystemRetryConfig.newBuilder().build();
     private Option<TimelineLayoutVersion> layoutVersion = Option.of(TimelineLayoutVersion.CURR_LAYOUT_VERSION);
@@ -685,8 +685,8 @@ public Builder setPayloadClassName(String payloadClassName) {
       return this;
     }
 
-    public Builder setMergerImpls(String mergerImpls) {
-      this.mergerImpls = mergerImpls;
+    public Builder setMergerStrategy(String mergerStrategy) {
+      this.mergerStrategy = mergerStrategy;
       return this;
     }
 
@@ -715,7 +715,7 @@ public HoodieTableMetaClient build() {
       ValidationUtils.checkArgument(basePath != null, "basePath needs to be set to init HoodieTableMetaClient");
       return newMetaClient(conf, basePath,
           loadActiveTimelineOnLoad, consistencyGuardConfig, layoutVersion, payloadClassName,
-          mergerImpls, fileSystemRetryConfig, props);
+          mergerStrategy, fileSystemRetryConfig, props);
     }
   }
 
@@ -732,7 +732,7 @@ public static class PropertyBuilder {
     private String recordKeyFields;
     private String archiveLogFolder;
     private String payloadClassName;
-    private String mergerImpls;
+    private String mergerStrategy;
     private Integer timelineLayoutVersion;
     private String baseFileFormat;
     private String preCombineField;
@@ -799,8 +799,8 @@ public PropertyBuilder setPayloadClassName(String payloadClassName) {
       return this;
     }
 
-    public PropertyBuilder setMergerImpls(String mergerImpls) {
-      this.mergerImpls = mergerImpls;
+    public PropertyBuilder setMergerStrategy(String mergerStrategy) {
+      this.mergerStrategy = mergerStrategy;
       return this;
     }
 
@@ -910,7 +910,7 @@ public PropertyBuilder fromMetaClient(HoodieTableMetaClient metaClient) {
           .setTableName(metaClient.getTableConfig().getTableName())
           .setArchiveLogFolder(metaClient.getArchivePath())
           .setPayloadClassName(metaClient.getTableConfig().getPayloadClass())
-          .setMergerImpls(metaClient.getTableConfig().getMergerImpls());
+          .setMergerStrategy(metaClient.getTableConfig().getMergerStrategy());
     }
 
     public PropertyBuilder fromProperties(Properties properties) {
@@ -940,9 +940,9 @@ public PropertyBuilder fromProperties(Properties properties) {
         setPayloadClassName(
             hoodieConfig.getString(HoodieTableConfig.PAYLOAD_CLASS_NAME));
       }
-      if (hoodieConfig.contains(HoodieTableConfig.MERGER_IMPLS)) {
-        setMergerImpls(
-            hoodieConfig.getString(HoodieTableConfig.MERGER_IMPLS));
+      if (hoodieConfig.contains(HoodieTableConfig.MERGER_STRATEGY)) {
+        setMergerStrategy(
+            hoodieConfig.getString(HoodieTableConfig.MERGER_STRATEGY));
       }
       if (hoodieConfig.contains(HoodieTableConfig.TIMELINE_LAYOUT_VERSION)) {
         setTimelineLayoutVersion(hoodieConfig.getInt(HoodieTableConfig.TIMELINE_LAYOUT_VERSION));
@@ -1021,8 +1021,8 @@ public Properties build() {
       if (tableType == HoodieTableType.MERGE_ON_READ && payloadClassName != null) {
         tableConfig.setValue(HoodieTableConfig.PAYLOAD_CLASS_NAME, payloadClassName);
       }
-      if (tableType == HoodieTableType.MERGE_ON_READ && mergerImpls != null) {
-        tableConfig.setValue(HoodieTableConfig.MERGER_IMPLS, mergerImpls);
+      if (tableType == HoodieTableType.MERGE_ON_READ && mergerStrategy != null) {
+        tableConfig.setValue(HoodieTableConfig.MERGER_STRATEGY, mergerStrategy);
       }
 
       if (null != tableCreateSchema) {

hudi-common/src/main/java/org/apache/hudi/common/util/HoodieRecordUtils.java

@@ -67,7 +67,7 @@ public static HoodieRecordMerger loadRecordMerger(String mergerClass) {
    * Instantiate a given class with a record merge.
    */
   public static HoodieRecordMerger generateRecordMerger(String basePath, EngineType engineType,
-      List<String> mergerClassList) {
+      List<String> mergerClassList, String mergerStrategy) {
     if (mergerClassList.isEmpty() || HoodieTableMetadata.isMetadataTable(basePath)) {
       return HoodieRecordUtils.loadRecordMerger(HoodieAvroRecordMerger.class.getName());
     } else {
@@ -81,6 +81,7 @@ public static HoodieRecordMerger generateRecordMerger(String basePath, EngineTyp
             }
           })
           .filter(Objects::nonNull)
+          .filter(merger -> merger.getMergingStrategy().equals(mergerStrategy))
           .filter(merger -> recordTypeCompatibleEngine(merger.getRecordType(), engineType))
           .findFirst()
           .orElse(HoodieRecordUtils.loadRecordMerger(HoodieAvroRecordMerger.class.getName()));

hudi-common/src/main/java/org/apache/hudi/common/util/StringUtils.java

@@ -33,6 +33,8 @@ public class StringUtils {
 
   public static final String EMPTY_STRING = "";
 
+  public static final String DEFAULT_MERGER_STRATEGY_UUID = "eeb8d96f-b1e4-49fd-bbf8-28ac514178e5";
+
   /**
    * <p>
    * Joins the elements of the provided array into a single String containing the provided list of elements.

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java

@@ -27,6 +27,7 @@
 import org.apache.hudi.common.model.HoodieCleaningPolicy;
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.config.HoodieIndexConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieException;
@@ -120,9 +121,16 @@ private FlinkOptions() {
       .key("record.merger.impls")
       .stringType()
       .defaultValue(HoodieAvroRecordMerger.class.getName())
-      .withDescription("List of HoodieMerger implementations constituting Hudi's merging strategy -- based on the engine used "
+      .withDescription("List of HoodieMerger implementations constituting Hudi's merging strategy -- based on the engine used. "
+          + "These merger impls will filter by record.merger.strategy. "
           + "Hudi will pick most efficient implementation to perform merging/combining of the records (during update, reading MOR table, etc)");
 
+  public static final ConfigOption<String> RECORD_MERGER_STRATEGY = ConfigOptions
+      .key("record.merger.strategy")
+      .stringType()
+      .defaultValue(StringUtils.DEFAULT_MERGER_STRATEGY_UUID)
+      .withDescription("Id of merger strategy.  Hudi will pick RecordMergers in record.merger.impls which has the same merger strategy id");
+
   public static final ConfigOption<String> PARTITION_DEFAULT_NAME = ConfigOptions
       .key("partition.default_name")
       .stringType()

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteFunction.java

@@ -18,9 +18,7 @@
 
 package org.apache.hudi.sink;
 
-import java.util.Arrays;
 import org.apache.hudi.client.WriteStatus;
-import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.model.HoodieAvroRecord;
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieOperation;
@@ -29,7 +27,6 @@
 import org.apache.hudi.common.model.HoodieRecordLocation;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.WriteOperationType;
-import org.apache.hudi.common.util.HoodieRecordUtils;
 import org.apache.hudi.common.util.ObjectSizeCalculator;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.configuration.FlinkOptions;
@@ -204,12 +201,7 @@ private void initWriteFunction() {
   }
 
   private void initMergeClass() {
-    List<String> mergers = Arrays.stream(config.getString(FlinkOptions.RECORD_MERGER_IMPLS).split(","))
-        .map(String::trim)
-        .distinct()
-        .collect(Collectors.toList());
-    recordMerger = HoodieRecordUtils.generateRecordMerger(writeClient.getConfig().getBasePath(), EngineType.FLINK,
-        mergers);
+    recordMerger = writeClient.getConfig().getRecordMerger();
     LOG.info("init hoodie merge with class [{}]", recordMerger.getClass().getName());
   }