[HUDI-8203] Make record merge mode the primary merging config

hudi-cli/src/main/java/org/apache/hudi/cli/commands/BootstrapCommand.java

@@ -72,8 +72,14 @@ public String bootstrap(
  help = "Class for Full bootstrap input provider") final String fullBootstrapInputProvider,
  @ShellOption(value = {"--schemaProviderClass"}, defaultValue = "",
  help = "SchemaProvider to attach schemas to bootstrap source data") final String schemaProviderClass,
- @ShellOption(value = {"--payloadClass"}, defaultValue = "org.apache.hudi.common.model.OverwriteWithLatestAvroPayload",
- help = "Payload Class") final String payloadClass,
+ @ShellOption(value = {"--payloadClass"}, defaultValue = "",
+ help = "Payload Class (deprecated). Use merge-mode for overwrite or event time merging.") final String payloadClass,
+ @ShellOption(value = {"--merge-mode", "--record-merge-mode"}, defaultValue = "",
+ help = "Merge mode to use. 'EVENT_TIME_ORDERING', 'OVERWRITE_WITH_LATEST', or 'CUSTOM' if you want to set a merge strategy") final String recordMergeMode,
+ @ShellOption(value = {"--merger-strategy", "--record-merger-strategy"}, defaultValue = "",
+ help = "Merge strategy to use. Only set when using 'CUSTOM' merge mode") final String recordMergerStrategy,
+ @ShellOption(value = {"--merger-impls"}, defaultValue = "",
+ help = "Comma separated list of classes that implement the record merger strategy") final String recordMergerImpls,
  @ShellOption(value = {"--parallelism"}, defaultValue = "1500", help = "Bootstrap writer parallelism") final int parallelism,
  @ShellOption(value = {"--sparkMaster"}, defaultValue = "", help = "Spark Master") String master,
  @ShellOption(value = {"--sparkMemory"}, defaultValue = "4G", help = "Spark executor memory") final String sparkMemory,
@@ -95,7 +101,8 @@ public String bootstrap(
 
  SparkMain.addAppArgs(sparkLauncher, SparkCommand.BOOTSTRAP, master, sparkMemory, tableName, tableType, targetPath, srcPath, rowKeyField,
  partitionPathField, String.valueOf(parallelism), schemaProviderClass, bootstrapIndexClass, selectorClass,
- keyGeneratorClass, fullBootstrapInputProvider, payloadClass, String.valueOf(enableHiveSync), propsFilePath);
+ keyGeneratorClass, fullBootstrapInputProvider, recordMergeMode, payloadClass, recordMergerStrategy, recordMergerImpls,
+ String.valueOf(enableHiveSync), propsFilePath);
  UtilHelpers.validateAndAddProperties(configs, sparkLauncher);
  Process process = sparkLauncher.launch();
  InputStreamConsumer.captureOutput(process);

hudi-cli/src/main/java/org/apache/hudi/cli/commands/SparkMain.java

@@ -23,6 +23,7 @@
 import org.apache.hudi.cli.utils.SparkUtil;
 import org.apache.hudi.client.SparkRDDWriteClient;
 import org.apache.hudi.client.common.HoodieSparkEngineContext;
+import org.apache.hudi.common.config.RecordMergeMode;
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
@@ -34,7 +35,6 @@
 import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.PartitionPathEncodeUtils;
-import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.config.HoodieBootstrapConfig;
 import org.apache.hudi.config.HoodieCleanConfig;
@@ -82,6 +82,7 @@
 import java.util.Locale;
 import java.util.Map;
 
+import static org.apache.hudi.common.util.StringUtils.isNullOrEmpty;
 import static org.apache.hudi.utilities.UtilHelpers.EXECUTE;
 import static org.apache.hudi.utilities.UtilHelpers.SCHEDULE;
 import static org.apache.hudi.utilities.UtilHelpers.SCHEDULE_AND_EXECUTE;
@@ -99,7 +100,7 @@ public class SparkMain {
  * Commands.
  */
  enum SparkCommand {
- BOOTSTRAP(18), ROLLBACK(6), DEDUPLICATE(8), ROLLBACK_TO_SAVEPOINT(6), SAVEPOINT(7),
+ BOOTSTRAP(21), ROLLBACK(6), DEDUPLICATE(8), ROLLBACK_TO_SAVEPOINT(6), SAVEPOINT(7),
  IMPORT(13), UPSERT(13), COMPACT_SCHEDULE(7), COMPACT_RUN(10), COMPACT_SCHEDULE_AND_EXECUTE(9),
  COMPACT_UNSCHEDULE_PLAN(9), COMPACT_UNSCHEDULE_FILE(10), COMPACT_VALIDATE(7), COMPACT_REPAIR(8),
  CLUSTERING_SCHEDULE(7), CLUSTERING_RUN(9), CLUSTERING_SCHEDULE_AND_EXECUTE(8), CLEAN(5),
@@ -129,7 +130,7 @@ List<String> makeConfigs(String[] args) {
  }
 
  String getPropsFilePath(String[] args) {
- return (args.length >= minArgsCount && !StringUtils.isNullOrEmpty(args[minArgsCount - 1]))
+ return (args.length >= minArgsCount && !isNullOrEmpty(args[minArgsCount - 1]))
  ? args[minArgsCount - 1] : null;
  }
  }
@@ -234,7 +235,7 @@ public static void main(String[] args) {
  break;
  case BOOTSTRAP:
  returnCode = doBootstrap(jsc, args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10],
- args[11], args[12], args[13], args[14], args[15], args[16], propsFilePath, configs);
+ args[11], args[12], args[13], args[14], args[15], args[16], args[17], args[18], args[19], propsFilePath, configs);
  break;
  case UPGRADE:
  case DOWNGRADE:
@@ -488,14 +489,15 @@ private static Map<String, String> getPropsForRewrite(HoodieTableMetaClient meta
  private static int doBootstrap(JavaSparkContext jsc, String tableName, String tableType, String basePath,
  String sourcePath, String recordKeyCols, String partitionFields, String parallelism, String schemaProviderClass,
  String bootstrapIndexClass, String selectorClass, String keyGenerator, String fullBootstrapInputProvider,
- String payloadClassName, String enableHiveSync, String propsFilePath, List<String> configs) throws IOException {
+ String recordMergeMode, String payloadClassName, String recordMergeStrategy, String recordMergerImpls,
+ String enableHiveSync, String propsFilePath, List<String> configs) throws IOException {
 
  TypedProperties properties = propsFilePath == null ? buildProperties(configs)
  : readConfig(jsc.hadoopConfiguration(), new Path(propsFilePath), configs).getProps(true);
 
  properties.setProperty(HoodieBootstrapConfig.BASE_PATH.key(), sourcePath);
 
- if (!StringUtils.isNullOrEmpty(keyGenerator) && KeyGeneratorType.getNames().contains(keyGenerator.toUpperCase(Locale.ROOT))) {
+ if (!isNullOrEmpty(keyGenerator) && KeyGeneratorType.getNames().contains(keyGenerator.toUpperCase(Locale.ROOT))) {
  properties.setProperty(HoodieWriteConfig.KEYGENERATOR_TYPE.key(), keyGenerator.toUpperCase(Locale.ROOT));
  } else {
  properties.setProperty(DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME().key(), keyGenerator);
@@ -514,6 +516,9 @@ private static int doBootstrap(JavaSparkContext jsc, String tableName, String ta
  cfg.schemaProviderClassName = schemaProviderClass;
  cfg.bootstrapIndexClass = bootstrapIndexClass;
  cfg.payloadClassName = payloadClassName;
+ cfg.recordMergeMode = RecordMergeMode.getValue(recordMergeMode);
+ cfg.recordMergerStrategy = recordMergeStrategy;
+ cfg.recordMergerImpls = recordMergerImpls;
  cfg.enableHiveSync = Boolean.valueOf(enableHiveSync);
 
  new BootstrapExecutor(cfg, jsc, HadoopFSUtils.getFs(basePath, jsc.hadoopConfiguration()),

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

@@ -23,6 +23,7 @@
 import org.apache.hudi.common.config.ConfigProperty;
 import org.apache.hudi.common.config.HoodieConfig;
 import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
+import org.apache.hudi.common.util.StringUtils;
 
 import java.io.File;
 import java.io.FileReader;
@@ -65,9 +66,6 @@ public class HoodiePayloadConfig extends HoodieConfig {
  .withDocumentation("Table column/field name to order records that have the same key, before "
  + "merging and writing to storage.");
 
- /** @deprecated Use {@link #PAYLOAD_CLASS_NAME} and its methods instead */
- @Deprecated
- public static final String DEFAULT_PAYLOAD_CLASS = PAYLOAD_CLASS_NAME.defaultValue();
  /** @deprecated Use {@link #PAYLOAD_CLASS_NAME} and its methods instead */
  @Deprecated
  public static final String PAYLOAD_CLASS_PROP = PAYLOAD_CLASS_NAME.key();
@@ -107,7 +105,9 @@ public Builder withPayloadEventTimeField(String payloadEventTimeField) {
  }
 
  public HoodiePayloadConfig.Builder withPayloadClass(String payloadClassName) {
- payloadConfig.setValue(PAYLOAD_CLASS_NAME, payloadClassName);
+ if (!StringUtils.isNullOrEmpty(payloadClassName)) {
+ payloadConfig.setValue(PAYLOAD_CLASS_NAME, payloadClassName);
+ }
  return this;
  }
 

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

@@ -35,18 +35,18 @@
 import org.apache.hudi.common.config.HoodieStorageConfig;
 import org.apache.hudi.common.config.HoodieTableServiceManagerConfig;
 import org.apache.hudi.common.config.HoodieTimeGeneratorConfig;
+import org.apache.hudi.common.config.RecordMergeMode;
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.fs.ConsistencyGuardConfig;
 import org.apache.hudi.common.fs.FileSystemRetryConfig;
-import org.apache.hudi.common.model.DefaultHoodieRecordPayload;
 import org.apache.hudi.common.model.HoodieAvroRecordMerger;
 import org.apache.hudi.common.model.HoodieCleaningPolicy;
 import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
 import org.apache.hudi.common.model.HoodieFileFormat;
 import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.model.HoodieTableType;
-import org.apache.hudi.common.model.RecordPayloadType;
 import org.apache.hudi.common.model.WriteConcurrencyMode;
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.table.HoodieTableConfig;
@@ -168,21 +168,21 @@ public class HoodieWriteConfig extends HoodieConfig {
 
  public static final ConfigProperty<String> WRITE_PAYLOAD_CLASS_NAME = ConfigProperty
  .key("hoodie.datasource.write.payload.class")
- .defaultValue(DefaultHoodieRecordPayload.class.getName())
+ .noDefaultValue()
  .markAdvanced()
+ .deprecatedAfter("1.0.0")
  .withDocumentation("Payload class used. Override this, if you like to roll your own merge logic, when upserting/inserting. "
  + "This will render any value set for PRECOMBINE_FIELD_OPT_VAL in-effective");
 
- public static final ConfigProperty<String> WRITE_PAYLOAD_TYPE = ConfigProperty
- .key("hoodie.datasource.write.payload.type")
- .defaultValue(RecordPayloadType.HOODIE_AVRO_DEFAULT.name())
- .markAdvanced()
+ public static final ConfigProperty<String> RECORD_MERGE_MODE = ConfigProperty
+ .key("hoodie.record.merge.mode")
+ .defaultValue(RecordMergeMode.EVENT_TIME_ORDERING.name())
  .sinceVersion("1.0.0")
- .withDocumentation(RecordPayloadType.class);
+ .withDocumentation(RecordMergeMode.class);
 
  public static final ConfigProperty<String> RECORD_MERGER_IMPLS = ConfigProperty
  .key("hoodie.datasource.write.record.merger.impls")
- .defaultValue(HoodieAvroRecordMerger.class.getName())
+ .noDefaultValue()
  .markAdvanced()
  .sinceVersion("0.13.0")
  .withDocumentation("List of HoodieMerger implementations constituting Hudi's merging strategy -- based on the engine used. "
@@ -191,7 +191,7 @@ public class HoodieWriteConfig extends HoodieConfig {
 
  public static final ConfigProperty<String> RECORD_MERGER_STRATEGY = ConfigProperty
  .key("hoodie.datasource.write.record.merger.strategy")
- .defaultValue(HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID)
+ .noDefaultValue()
  .markAdvanced()
  .sinceVersion("0.13.0")
  .withDocumentation("Id of merger strategy. Hudi will pick HoodieRecordMerger implementations in hoodie.datasource.write.record.merger.impls which has the same merger strategy id");
@@ -844,11 +844,6 @@ public class HoodieWriteConfig extends HoodieConfig {
  */
  @Deprecated
  public static final String WRITE_PAYLOAD_CLASS = WRITE_PAYLOAD_CLASS_NAME.key();
- /**
- * @deprecated Use {@link #WRITE_PAYLOAD_CLASS_NAME} and its methods instead
- */
- @Deprecated
- public static final String DEFAULT_WRITE_PAYLOAD_CLASS = WRITE_PAYLOAD_CLASS_NAME.defaultValue();
  /**
  * @deprecated Use {@link #KEYGENERATOR_CLASS_NAME} and its methods instead
  */
@@ -1239,16 +1234,69 @@ public String getBasePath() {
  }
 
  public HoodieFileFormat getBaseFileFormat() {
- return HoodieFileFormat.valueOf(getStringOrDefault(BASE_FILE_FORMAT));
+ return HoodieFileFormat.valueOf(getStringOrDefault(BASE_FILE_FORMAT).toUpperCase());
+ }
+
+ public String getRecordMergerStrategy() {
+ return getString(RECORD_MERGER_STRATEGY);
+ }
+
+ public RecordMergeMode getRecordMergeMode() {
+ return RecordMergeMode.getValue(getString(RECORD_MERGE_MODE));
  }
 
  public HoodieRecordMerger getRecordMerger() {
- List<String> mergers = StringUtils.split(getStringOrDefault(RECORD_MERGER_IMPLS), ",").stream()
+ List<String> mergers = StringUtils.split(getString(RECORD_MERGER_IMPLS), ",").stream()
  .map(String::trim)
  .distinct()
  .collect(Collectors.toList());
- String recordMergerStrategy = getString(RECORD_MERGER_STRATEGY);
- return HoodieRecordUtils.createRecordMerger(getString(BASE_PATH), engineType, mergers, recordMergerStrategy);
+ return getRecordMerger(getString(BASE_PATH), getRecordMergeMode(),
+ engineType, getLogDataBlockFormat(), mergers, getStringOpt(RECORD_MERGER_STRATEGY));
+ }
+
+ public static HoodieRecordMerger getRecordMerger(String basePath,
+ RecordMergeMode mergeMode,
+ EngineType engineType,
+ HoodieLogBlock.HoodieLogBlockType logBlockType,
+ List<String> mergers,
+ Option<String> strategy) {
+ switch (logBlockType) {
+ case HFILE_DATA_BLOCK:
+ return HoodieAvroRecordMerger.INSTANCE;
+ case AVRO_DATA_BLOCK:
+ case PARQUET_DATA_BLOCK:
+ //TODO: [HUDI-8317] return getMergerByMode(basePath, mergeMode, engineType, mergers, strategy);
+ return HoodieRecordUtils.createRecordMerger(basePath, engineType, mergers, strategy);
+ default:
+ throw new IllegalStateException("This log block type is not implemented");
+ }
+ }
+
+ private static HoodieRecordMerger getMergerByMode(String basePath,
+ RecordMergeMode mergeMode,
+ EngineType engineType,
+ List<String> mergers,
+ Option<String> strategy) {
+ //TODO: [HUDI-8202] make this custom mergers only
+ switch (mergeMode) {
+ case EVENT_TIME_ORDERING:
+ switch (engineType) {
+ case SPARK:
+ return HoodieRecordUtils.loadRecordMerger("org.apache.hudi.DefaultSparkRecordMerger");
+ default:
+ return HoodieRecordUtils.createRecordMerger(basePath, engineType, mergers, strategy);
+ }
+ case OVERWRITE_WITH_LATEST:
+ switch (engineType) {
+ case SPARK:
+ return HoodieRecordUtils.loadRecordMerger("org.apache.hudi.OverwriteWithLatestSparkRecordMerger");
+ default:
+ return HoodieRecordUtils.createRecordMerger(basePath, engineType, mergers, strategy);
+ }
+ case CUSTOM:
+ default:
+ return HoodieRecordUtils.createRecordMerger(basePath, engineType, mergers, strategy);
+ }
  }
 
  public String getSchema() {
@@ -1746,6 +1794,10 @@ public int getAsyncClusterMaxCommits() {
  return getInt(HoodieClusteringConfig.ASYNC_CLUSTERING_MAX_COMMITS);
  }
 
+ public String getAvroPayloadClass() {
+ return getStringOpt(HoodiePayloadConfig.PAYLOAD_CLASS_NAME).orElseGet(() -> HoodieRecordPayload.getAvroPayloadForMergeMode(getRecordMergeMode()));
+ }
+
  public String getPayloadClass() {
  return getString(HoodiePayloadConfig.PAYLOAD_CLASS_NAME);
  }
@@ -2200,9 +2252,8 @@ public boolean parquetBloomFilterEnabled() {
  return getBooleanOrDefault(HoodieStorageConfig.PARQUET_WITH_BLOOM_FILTER_ENABLED);
  }
 
- public Option<HoodieLogBlock.HoodieLogBlockType> getLogDataBlockFormat() {
- return Option.ofNullable(getString(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT))
- .map(HoodieLogBlock.HoodieLogBlockType::fromId);
+ public HoodieLogBlock.HoodieLogBlockType getLogDataBlockFormat() {
+ return HoodieLogBlock.inferLogBlockWriteFormat(getString(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT), getBaseFileFormat());
  }
 
  public long getLogFileMaxSize() {
@@ -2903,12 +2954,16 @@ public Builder withWritePayLoad(String payload) {
  }
 
  public Builder withRecordMergerImpls(String recordMergerImpls) {
- writeConfig.setValue(RECORD_MERGER_IMPLS, recordMergerImpls);
+ if (!StringUtils.isNullOrEmpty(recordMergerImpls)) {
+ writeConfig.setValue(RECORD_MERGER_IMPLS, recordMergerImpls);
+ }
  return this;
  }
 
  public Builder withRecordMergerStrategy(String recordMergerStrategy) {
- writeConfig.setValue(RECORD_MERGER_STRATEGY, recordMergerStrategy);
+ if (recordMergerStrategy != null) {
+ writeConfig.setValue(RECORD_MERGER_STRATEGY, recordMergerStrategy);
+ }
  return this;
  }
 
@@ -3099,6 +3154,13 @@ public Builder withPayloadConfig(HoodiePayloadConfig payloadConfig) {
  return this;
  }
 
+ public Builder withRecordMergeMode(RecordMergeMode recordMergeMode) {
+ if (recordMergeMode != null) {
+ writeConfig.setValue(RECORD_MERGE_MODE, recordMergeMode.name());
+ }
+ return this;
+ }
+
  public Builder withMetadataConfig(HoodieMetadataConfig metadataConfig) {
  writeConfig.getProps().putAll(metadataConfig.getProps());
  isMetadataConfigSet = true;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/index/HoodieIndexUtils.java

@@ -257,11 +257,11 @@ private static <R> HoodieData<HoodieRecord<R>> getExistingRecords(
  * evaluates the merge.
  */
  private static Pair<HoodieWriteConfig, Option<BaseKeyGenerator>> getKeygenAndUpdatedWriteConfig(HoodieWriteConfig config, HoodieTableConfig tableConfig) {
- if (config.getPayloadClass().equals("org.apache.spark.sql.hudi.command.payload.ExpressionPayload")) {
+ if (config.getAvroPayloadClass().equals("org.apache.spark.sql.hudi.command.payload.ExpressionPayload")) {
  TypedProperties typedProperties = new TypedProperties(config.getProps());
  // set the payload class to table's payload class and not expresison payload. this will be used to read the existing records
- typedProperties.setProperty(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key(), tableConfig.getPayloadClass());
- typedProperties.setProperty(HoodieTableConfig.PAYLOAD_CLASS_NAME.key(), tableConfig.getPayloadClass());
+ typedProperties.setProperty(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key(), tableConfig.getAvroPayloadClass());
+ typedProperties.setProperty(HoodieTableConfig.PAYLOAD_CLASS_NAME.key(), tableConfig.getAvroPayloadClass());
  HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder().withProperties(typedProperties).build();
  try {
  return Pair.of(writeConfig, Option.of((BaseKeyGenerator) HoodieAvroKeyGeneratorFactory.createKeyGenerator(writeConfig.getProps())));