yihua commented on code in PR #11943:
URL: https://github.com/apache/hudi/pull/11943#discussion_r1805616264
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -371,9 +365,9 @@ public HoodieTableConfig(HoodieStorage storage, StoragePath
metaPath, String pay
setValue(PAYLOAD_CLASS_NAME, payloadClassName);
needStore = true;
}
- if (contains(PAYLOAD_TYPE) && payloadClassName != null
- &&
!payloadClassName.equals(RecordPayloadType.valueOf(getString(PAYLOAD_TYPE)).getClassName()))
{
- setValue(PAYLOAD_TYPE,
RecordPayloadType.fromClassName(payloadClassName).name());
+ if (contains(RECORD_MERGE_MODE) && recordMergeMode != null
Review Comment:
Should we throw an error here since it's going to change the record merge
mode? Similar for the payload class as we are not going to allow payload class
to change in the table configs (unless through a custom tooling).
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -698,7 +686,59 @@ public String getPayloadClass() {
* Read the payload class for HoodieRecords from the table properties.
*/
public String getRecordMergerStrategy() {
- return getStringOrDefault(RECORD_MERGER_STRATEGY);
+ return getString(RECORD_MERGER_STRATEGY);
+ }
+
+ public static Triple<RecordMergeMode, String, String>
inferCorrectMergingBehavior(RecordMergeMode recordMergeMode, String
payloadClassName, String recordMergerStrategy) {
+ RecordMergeMode inferRecordMergeMode;
+ String inferPayloadClassName;
+ String inferRecordMergerStrategy;
+
+ if (isNullOrEmpty(payloadClassName)) {
+ if (isNullOrEmpty(recordMergerStrategy)) {
+ checkArgument(recordMergeMode != RecordMergeMode.CUSTOM, "Custom merge
mode should only be used if you set a merge strategy");
+ if (recordMergeMode == null) {
+ inferRecordMergeMode = RECORD_MERGE_MODE.defaultValue();
+ } else {
+ inferRecordMergeMode = recordMergeMode;
+ }
+ inferRecordMergerStrategy =
HoodieRecordMerger.getAvroMergerStrategyFromMergeMode(inferRecordMergeMode);
Review Comment:
If `HoodieRecordMerger.getAvroMergerStrategyFromMergeMode` is only used
here, it's better to embed to switch-case logic here directly for clarity.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -185,41 +196,24 @@ public class HoodieTableConfig extends HoodieConfig {
.key("hoodie.timeline.layout.version")
.noDefaultValue()
.withDocumentation("Version of timeline used, by the table.");
-
- //TODO: why is this the default? not OVERWRITE_WITH_LATEST?
- public static final ConfigProperty<String> RECORD_MERGE_MODE = ConfigProperty
+
+ public static final ConfigProperty<RecordMergeMode> RECORD_MERGE_MODE =
ConfigProperty
.key("hoodie.record.merge.mode")
- .defaultValue(RecordMergeMode.EVENT_TIME_ORDERING.name())
+ .defaultValue(RecordMergeMode.EVENT_TIME_ORDERING)
.sinceVersion("1.0.0")
.withDocumentation(RecordMergeMode.class);
public static final ConfigProperty<String> PAYLOAD_CLASS_NAME =
ConfigProperty
.key("hoodie.compaction.payload.class")
- .defaultValue(DefaultHoodieRecordPayload.class.getName())
+ .noDefaultValue()
.deprecatedAfter("1.0.0")
.withDocumentation("Payload class to use for performing merges,
compactions, i.e merge delta logs with current base file and then "
+ " produce a new base file.");
- public static final ConfigProperty<String> PAYLOAD_TYPE = ConfigProperty
- .key("hoodie.compaction.payload.type")
- .defaultValue(RecordPayloadType.HOODIE_AVRO_DEFAULT.name())
- .sinceVersion("1.0.0")
- .withDocumentation(RecordPayloadType.class);
-
public static final ConfigProperty<String> RECORD_MERGER_STRATEGY =
ConfigProperty
- .key("hoodie.compaction.record.merger.strategy")
- .defaultValue(HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID)
- .withInferFunction(cfg -> {
- switch
(RecordMergeMode.valueOf(cfg.getStringOrDefault(RECORD_MERGE_MODE))) {
- case EVENT_TIME_ORDERING:
- return Option.of(HoodieRecordMerger.DEFAULT_MERGER_STRATEGY_UUID);
- case OVERWRITE_WITH_LATEST:
- return
Option.of(HoodieRecordMerger.OVERWRITE_MERGER_STRATEGY_UUID);
- case CUSTOM:
- default:
- return Option.empty();
- }
- })
+ .key("hoodie.record.merge.custom.strategy")
Review Comment:
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:
##########
@@ -698,7 +686,59 @@ public String getPayloadClass() {
* Read the payload class for HoodieRecords from the table properties.
*/
public String getRecordMergerStrategy() {
- return getStringOrDefault(RECORD_MERGER_STRATEGY);
+ return getString(RECORD_MERGER_STRATEGY);
+ }
+
+ public static Triple<RecordMergeMode, String, String>
inferCorrectMergingBehavior(RecordMergeMode recordMergeMode, String
payloadClassName, String recordMergerStrategy) {
+ RecordMergeMode inferRecordMergeMode;
+ String inferPayloadClassName;
+ String inferRecordMergerStrategy;
+
+ if (isNullOrEmpty(payloadClassName)) {
+ if (isNullOrEmpty(recordMergerStrategy)) {
+ checkArgument(recordMergeMode != RecordMergeMode.CUSTOM, "Custom merge
mode should only be used if you set a merge strategy");
Review Comment:
If would be helpful to add a line of docs to explain each scenario, e.g.,
`// No payload class name, no record merge strategy`.
##########
hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableConfig.java:
##########
@@ -698,7 +686,59 @@ public String getPayloadClass() {
* Read the payload class for HoodieRecords from the table properties.
*/
public String getRecordMergerStrategy() {
- return getStringOrDefault(RECORD_MERGER_STRATEGY);
+ return getString(RECORD_MERGER_STRATEGY);
+ }
+
+ public static Triple<RecordMergeMode, String, String>
inferCorrectMergingBehavior(RecordMergeMode recordMergeMode, String
payloadClassName, String recordMergerStrategy) {
+ RecordMergeMode inferRecordMergeMode;
+ String inferPayloadClassName;
+ String inferRecordMergerStrategy;
+
+ if (isNullOrEmpty(payloadClassName)) {
+ if (isNullOrEmpty(recordMergerStrategy)) {
+ checkArgument(recordMergeMode != RecordMergeMode.CUSTOM, "Custom merge
mode should only be used if you set a merge strategy");
+ if (recordMergeMode == null) {
+ inferRecordMergeMode = RECORD_MERGE_MODE.defaultValue();
+ } else {
+ inferRecordMergeMode = recordMergeMode;
+ }
+ inferRecordMergerStrategy =
HoodieRecordMerger.getAvroMergerStrategyFromMergeMode(inferRecordMergeMode);
+ } else {
+ inferRecordMergerStrategy = recordMergerStrategy;
+ if (recordMergerStrategy.equals(DEFAULT_MERGER_STRATEGY_UUID)) {
+ inferRecordMergeMode = EVENT_TIME_ORDERING;
+ } else if
(recordMergerStrategy.equals(OVERWRITE_MERGER_STRATEGY_UUID)) {
+ inferRecordMergeMode = OVERWRITE_WITH_LATEST;
+ } else {
+
checkArgument(!recordMergerStrategy.equals(PAYLOAD_BASED_MERGER_STRATEGY_UUID),
"Payload based strategy should only be used if you have a custom payload");
+ inferRecordMergeMode = CUSTOM;
+ }
+ }
+ inferPayloadClassName =
HoodieRecordPayload.getAvroPayloadForMergeMode(inferRecordMergeMode);
+ } else {
+ inferPayloadClassName = payloadClassName;
+ if (payloadClassName.equals(DefaultHoodieRecordPayload.class.getName()))
{
+ //TODO: after mergers are safe to use on the write side, add this
check and get rid of the if
+ // checkArgument(isNullOrEmpty(recordMergerStrategy) ||
recordMergerStrategy.equals(DEFAULT_MERGER_STRATEGY_UUID), "Record merge
strategy cannot be set if a merge payload is used");
+ if (isNullOrEmpty(recordMergerStrategy) ||
recordMergerStrategy.equals(DEFAULT_MERGER_STRATEGY_UUID)) {
+ inferRecordMergeMode = EVENT_TIME_ORDERING;
+ inferRecordMergerStrategy = DEFAULT_MERGER_STRATEGY_UUID;
+ } else {
+ inferRecordMergeMode = CUSTOM;
+ inferRecordMergerStrategy = recordMergerStrategy;
+ }
+ } else if
(payloadClassName.equals(OverwriteWithLatestAvroPayload.class.getName())) {
+ checkArgument(isNullOrEmpty(recordMergerStrategy) ||
recordMergerStrategy.equals(OVERWRITE_MERGER_STRATEGY_UUID), "Record merge
strategy cannot be set if a merge payload is used");
+ inferRecordMergeMode = OVERWRITE_WITH_LATEST;
+ inferRecordMergerStrategy = OVERWRITE_MERGER_STRATEGY_UUID;
Review Comment:
For merge modes of `OVERWRITE_WITH_LATEST` and `EVENT_TIME_ORDERING`, should
we avoid persisting the record merge strategy ID?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
