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

jonvex · 2024-09-13T21:16:54Z

Change Logs

Record Merge Mode is the primary way to configure how your merging logic works. Avro payload based merging is only used if you set a payload that is not the default payload or the overwrite with latest payload.

The filegroup reader now will use payload based merging when the table is configured to use a custom avro payload.

Fg reader now uses payload based merging if the user uses that on the write side. This leads to consistent results when reading vs compaction

Impact

Part of umbrella to simplify merge configs for 1.X

Risk level (write none, low medium or high below)

low

Documentation Update

N/A

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

…load

…verwrite

…t for the payload. And refactor create avro record to move it to the base buffer instead of the reader context

yihua · 2024-10-17T21:40:01Z

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

- /** @deprecated Use {@link #PAYLOAD_CLASS_NAME} and its methods instead */
- @Deprecated
- public static final String DEFAULT_PAYLOAD_CLASS = PAYLOAD_CLASS_NAME.defaultValue();


Regarding this variable for backwards compatible read (https:/apache/hudi/pull/11943/files#r1805475755), could we hard code it to be DefaultHoodieRecordPayload.class.getName()?

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

yihua · 2024-10-17T21:42:24Z

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

- .markAdvanced()
+ public static final ConfigProperty<String> RECORD_MERGE_MODE = ConfigProperty
+ .key("hoodie.datasource.write.record.merge.mode")
+ .defaultValue(RecordMergeMode.EVENT_TIME_ORDERING.name())


I remember that this can be switched back to OVERWRITE_WITH_LATEST. Is there still an issue? I'm OK with EVENT_TIME_ORDERING; trying to see if it still strictly requires precombine/ordering field to be set.

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

yihua · 2024-10-17T21:46:56Z

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

+ 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 = "",


We can make all record merge configs to start with --record-merge-, hoodie.write.record.merge., ``hoodie.table.record.merge., e.g., --record-merge-strategy`, `hoodie.write.record.merge.strategy`, to be consistent.

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

yihua · 2024-10-17T21:48:32Z

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

- * @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();


We can hardcode the legacy default value here.

yihua · 2024-10-17T21:49:58Z

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

@@ -1239,16 +1233,19 @@ public String getBasePath() {
 }

 public HoodieFileFormat getBaseFileFormat() {
- return HoodieFileFormat.valueOf(getStringOrDefault(BASE_FILE_FORMAT));
+ return HoodieFileFormat.valueOf(getStringOrDefault(BASE_FILE_FORMAT).toUpperCase());


nit: could we handle the handling of case sensitivity inside HoodieFileFormat?

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

yihua · 2024-10-17T22:02:05Z

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

- .collect(Collectors.toList());
- String recordMergerStrategy = getString(RECORD_MERGER_STRATEGY);
- return HoodieRecordUtils.createRecordMerger(getString(BASE_PATH), engineType, mergers, recordMergerStrategy);
+ return HoodieRecordUtils.createRecordMerger(getString(BASE_PATH), engineType, getString(RECORD_MERGER_IMPLS), getString(RECORD_MERGER_STRATEGY));


Is this API used by Avro merger or custom merge mode only?

It is used to get the avro merger as well

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

yihua · 2024-10-17T22:05:53Z

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

 } else if (config.contains("hoodie.datasource.write.payload.class")) {
 payloadClassName = config.getString("hoodie.datasource.write.payload.class");
 } else {
- payloadClassName = RecordPayloadType.valueOf(PAYLOAD_TYPE.defaultValue()).getClassName();
+ return DefaultHoodieRecordPayload.class.getName();


Use the defined default legacy payload class variable here instead of hardcoding the value.

yihua · 2024-10-17T22:09:36Z

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

@@ -1747,7 +1744,7 @@ public int getAsyncClusterMaxCommits() {
 }

 public String getPayloadClass() {
- return getString(HoodiePayloadConfig.PAYLOAD_CLASS_NAME);
+ return RecordPayloadType.getPayloadClassName(this);


Have you checked if all existing logic of getting the payload class go through this method? Since this logic now returns the default payload class from the method instead of the config default value, we need to make sure any logic using getString(HoodiePayloadConfig.PAYLOAD_CLASS_NAME) should migrate to this API too, to avoid possible null or empty payload class name.

yihua · 2024-10-17T22:18:24Z

...ient/hudi-client-common/src/main/java/org/apache/hudi/metadata/HoodieMetadataWriteUtils.java

+ .withReleaseResourceEnabled(writeConfig.areReleaseResourceEnabled())
+ .withRecordMergeMode(RecordMergeMode.CUSTOM)
+ .withRecordMergerStrategy(HoodieRecordMerger.PAYLOAD_BASED_MERGER_STRATEGY_UUDID)
+ .withPayloadConfig(HoodiePayloadConfig.newBuilder()


Is it because the payload type is set before (HOODIE_METADATA), which is removed now?

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

...nt/hudi-spark-client/src/main/java/org/apache/hudi/OverwriteWithLatestSparkRecordMerger.java

...ient/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java

yihua · 2024-10-17T22:43:31Z

...di-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala

@@ -127,6 +129,14 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
 deserializer.deserialize(avroRecord).get.asInstanceOf[InternalRow]
 }

+ override def convertToAvroRecord(record: InternalRow, schema: Schema): GenericRecord = {


Note to myself: revisit for integration and performance impact.

hudi-common/src/main/java/org/apache/hudi/common/config/HoodieCommonConfig.java

hudi-common/src/main/java/org/apache/hudi/common/config/HoodieConfig.java

yihua · 2024-10-17T22:52:45Z

hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java

@@ -55,7 +57,7 @@ public abstract class HoodieReaderContext<T> {
 private HoodieFileGroupReaderSchemaHandler<T> schemaHandler = null;
 private String tablePath = null;
 private String latestCommitTime = null;
- private HoodieRecordMerger recordMerger = null;
+ private Option<HoodieRecordMerger> recordMerger = null;


Option.empty()?

We want to make sure it is set. If we get an npe then we know there is a bug. If we have it as option.empty then it is possible that the bug will happen silently and cause correctness issues

yihua · 2024-10-17T22:54:11Z

hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java

+ public void setRecordMerger(Option<HoodieRecordMerger> recordMerger) {
 this.recordMerger = recordMerger;
 }


Should the argument avoid Option?

Suggested change

public void setRecordMerger(Option<HoodieRecordMerger> recordMerger) {

this.recordMerger = recordMerger;

}

public void setRecordMerger(HoodieRecordMerger recordMerger) {

this.recordMerger = Option.of(recordMerger);

}

yihua · 2024-10-17T22:57:14Z

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

+ return persistedOrderingVal == null || ((persistedOrderingVal instanceof String || incomingOrderingVal instanceof String)
+ ? ((Comparable) persistedOrderingVal.toString()).compareTo(incomingOrderingVal.toString()) <= 0
+ : ((Comparable) persistedOrderingVal).compareTo(incomingOrderingVal) <= 0);


To be safe, let's avoid any changes in existing payload classes.

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

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

…tColumnProjection) and Test Call repair_overwrite_hoodie_props Procedure

yihua · 2024-10-17T23:17:12Z

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

+ static String getAvroPayloadForMergeMode(RecordMergeMode mergeMode) {
+ switch (mergeMode) {
+ //TODO: After we have merge mode working for writing, we should have a dummy payload that will throw exception when used
+ default:


Should CUSTOM mode be handled too? Is it easier to store the payload class name in the RecordMergeMode enum itself?

CUSTOM doesn't have an associated payload class. For now we will default to the default payload. But once we fix the experience of using engine native records in the writer, we should not be using payloads anymore.

yihua · 2024-10-18T00:07:25Z

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

- return Option.empty();
- }
- })
+ .key("hoodie.record.merge.custom.strategy")


Let's make sure the config naming is consistent among all write and table configs for merging behavior.

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

yihua · 2024-10-18T00:37:36Z

hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReader.java

- this.recordMerger = readerContext.getRecordMerger(tableConfig.getRecordMergerStrategy());
- readerContext.setRecordMerger(this.recordMerger);
+ readerContext.setRecordMerger(readerContext.getRecordMerger(tableConfig.getRecordMergeMode(),
+  tableConfig.getRecordMergerStrategy(), props.getString("hoodie.datasource.write.record.merger.impl","")));


Reference the config property declaration of "hoodie.datasource.write.record.merger.impl".

Could we directly do readerContext.setRecordMerger(tableConfig.getRecordMergeMode(), tableConfig.getRecordMergerStrategy(), props.getString("hoodie.datasource.write.record.merger.impl","")) by changing the setter logic?

It's good that all immutable merge configs come from table config now. Should we persist hoodie.datasource.write.record.merger.impl in table config too and make it mutable, so it also comes from the table config?

yihua · 2024-10-18T00:56:51Z

...mmon/src/main/java/org/apache/hudi/common/table/read/HoodieFileGroupReaderSchemaHandler.java

@@ -162,6 +165,31 @@ private Schema generateRequiredSchema() {
 return appendFieldsToSchemaDedupNested(requestedSchema, addedFields);
 }

+ private static String[] getMandatoryFieldsForMerging(HoodieTableConfig cfg, Option<HoodieRecordMerger> recordMerger) {
+ if (cfg.getRecordMergeMode() == RecordMergeMode.CUSTOM) {
+ return recordMerger.get().getMandatoryFieldsForMerging(cfg);


Should the default implementation of getMandatoryFieldsForMerging return all columns since custom merging may need all columns regardless of the request schema (to be followed up in a separate fix)? Only the DefaultSparkRecordMerger and OverwriteWithLatestSparkRecordMerger(and counterpart classes for other engines) should overwrite the implementation to return record key and optional preCombine field.

...rc/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedFileGroupRecordBuffer.java

yihua · 2024-10-18T01:01:36Z

...rc/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedFileGroupRecordBuffer.java

@@ -199,7 +199,7 @@ public void processNextDeletedRecord(DeleteRecord deleteRecord, Serializable rec
 if (recordOpt.isPresent()) {
 String recordKey = recordOpt.get().getRecordKey();
 records.put(recordPosition, Pair.of(Option.empty(), readerContext.generateMetadataForRecord(
- recordKey, recordOpt.get().getPartitionPath(), recordOpt.get().getOrderingValue(), orderingFieldType)));
+ recordKey, recordOpt.get().getPartitionPath(), recordOpt.get().getOrderingValue() == null ? orderingFieldDefault : recordOpt.get().getOrderingValue(), orderingFieldTypeOpt)));


We should not set orderingFieldDefault which can cause confusion and written to the storage eventually.

then what should we use as the ordering value when it is null?

yihua · 2024-10-18T01:06:28Z

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

@@ -71,6 +71,11 @@ public static HoodieRecordMerger loadRecordMerger(String mergerClass) {
 }
 }

+ public static HoodieRecordMerger createRecordMerger(String basePath, EngineType engineType,
+ String mergerImpls, String recordMergerStrategy) {
+ return createRecordMerger(basePath, engineType, ConfigUtils.split2List(mergerImpls), recordMergerStrategy);


Should HoodieTableConfig parse the custom merge implementation class list, i.e., HoodieTableConfig#getRecordMergerStrategy returns the List<String> mergerClassList, instead of passing the config value all the way down? and get rid of this new API.

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java

hudi-bot · 2024-10-20T13:58:56Z

CI report:

3c8df12 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

Jonathan Vexler added 2 commits September 13, 2024 17:16

first attempt at making merge mode the primary config for merging/pay…

a010515

…load

remove unused methods in meta client

d3045da

github-actions bot added the size:L PR with lines of changes in (300, 1000] label Sep 13, 2024

Jonathan Vexler added 2 commits September 16, 2024 14:52

Merge branch 'master' into create_new_merger_configs

371fd96

get the quickstart to pass

1a64f42

jonvex changed the title ~~first attempt at making merge mode the primary config for merging/pay…~~ [HUDI-8203] Make record merge mode the primary merging config Sep 16, 2024

Jonathan Vexler added 2 commits September 16, 2024 17:51

get rid of defaults for payload in hoodie option config

373f67f

don't infer based on engine type now

8745ea0

github-actions bot added size:XL PR with lines of changes > 1000 and removed size:L PR with lines of changes in (300, 1000] labels Sep 18, 2024

Jonathan Vexler added 20 commits September 18, 2024 19:48

Merge branch 'master' into create_new_merger_configs

f77cb87

fix some things and fix the issue with MIT

60755d4

fix checkstyle

aac457a

support avro read strategy

63c9a00

checkstyle

c82c27f

allow payload during read

5d466c3

Merge branch 'master' into create_new_merger_configs

416835f

fix merge issue

9795fe8

validate that we don't try to do payload merging with event time or o…

ab6d567

…verwrite

fix bugs and tests

ae29647

fix failing tests

12e8de2

fix test

ce04bd5

fix testMergerStrategySet

01f2024

fix utf8 comparable issue in default record payload

ffc94b8

fix getting payload in bootstrap

c789678

fix testCreateHoodieConfigWithAsyncClustering

88799a2

fix log format capitalization

638d350

deprecate payload config

3e486d2

fix checkstyle and make some more changes like not calling get defaul…

1db1fec

…t for the payload. And refactor create avro record to move it to the base buffer instead of the reader context

infer merge mode if they use default or overwrite payload

8d0d358

yihua reviewed Oct 17, 2024

View reviewed changes

fix some more tests

25897cb

yihua reviewed Oct 17, 2024

View reviewed changes

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordMerger.java Outdated Show resolved Hide resolved

hudi-common/src/main/java/org/apache/hudi/common/model/HoodieRecordMerger.java Outdated Show resolved Hide resolved

fix testMergeOnReadSnapshotRelationWithDeltaLogsFallback()(TestParque…

9119372

…tColumnProjection) and Test Call repair_overwrite_hoodie_props Procedure

yihua reviewed Oct 17, 2024

View reviewed changes

yihua reviewed Oct 18, 2024

View reviewed changes

fix failing tests

addd352

yihua reviewed Oct 18, 2024

View reviewed changes

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java Show resolved Hide resolved

hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java Show resolved Hide resolved

Jonathan Vexler added 12 commits October 18, 2024 10:24

fix azure tests

4cdbd16

add more validation to infer merge mode

f619f0b

address review comments and fix tests

f55704f

fix checkstyle

ad0682a

remove duplicate test case

4d11955

fix style

b851357

throw exception if no valid merger impl is present for custom merger

4303223

address review comments and fix tests

9a56f78

address more review comments

08b73c0

revert config name change

1bb38d2

fix failing tests

413ebc4

change config name to have id in it, but not table config

75beb3d

apache deleted a comment from hudi-bot Oct 18, 2024

Merge branch 'master' into create_new_merger_configs

3c8df12

jonvex requested a review from yihua October 20, 2024 00:49

apache deleted a comment from hudi-bot Oct 20, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

jonvex commented Sep 13, 2024 •

edited

Loading

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

jonvex Oct 18, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

jonvex Oct 18, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

yihua Oct 17, 2024

jonvex Oct 18, 2024

yihua Oct 18, 2024

yihua Oct 18, 2024

yihua Oct 18, 2024

yihua Oct 18, 2024

yihua Oct 18, 2024

yihua Oct 18, 2024

jonvex Oct 18, 2024

yihua Oct 18, 2024

hudi-bot commented Oct 20, 2024

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

Are you sure you want to change the base?

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

Conversation

jonvex commented Sep 13, 2024 • edited Loading

Change Logs

Impact

Risk level (write none, low medium or high below)

Documentation Update

Contributor's checklist

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hudi-bot commented Oct 20, 2024

CI report:

jonvex commented Sep 13, 2024 •

edited

Loading