[HUDI-6961] Fix deletes with custom delete field in DefaultHoodieRecordPayload #9892

yihua · 2023-10-19T21:32:10Z

Change Logs

This PR fixes the DefaultHoodieRecordPayload to allow the records with custom delete key (hoodie.payload.delete.field) and delete marker (hoodie.payload.delete.marker) to be properly ingested. Before this fix, the write fails with the following exception

Error for key:HoodieKey { recordKey=0 partitionPath=} is 
java.util.NoSuchElementException: No value present in Option
	at org.apache.hudi.common.util.Option.get(Option.java:89)
	at org.apache.hudi.common.model.HoodieAvroRecord.prependMetaFields(HoodieAvroRecord.java:132)
	at org.apache.hudi.io.HoodieCreateHandle.doWrite(HoodieCreateHandle.java:144)
	at org.apache.hudi.io.HoodieWriteHandle.write(HoodieWriteHandle.java:180)
	at org.apache.hudi.execution.CopyOnWriteInsertHandler.consume(CopyOnWriteInsertHandler.java:98)
	at org.apache.hudi.execution.CopyOnWriteInsertHandler.consume(CopyOnWriteInsertHandler.java:42)
	at org.apache.hudi.common.util.queue.SimpleExecutor.execute(SimpleExecutor.java:69)

The bug is introduced by the RFC-46 implementation and the refactoring on payload merging.

To fix the issue, the following changes are made:

After RFC-46, whether a record is a delete or not is determined at the instantiation of the record payload. For DefaultHoodieRecordPayload, this means that the configuration of delete key (hoodie.payload.delete.field) and delete marker (hoodie.payload.delete.marker) must be passed to the constructor. As the record payload (HoodieRecordPayload) creation is realized through Java reflection, the signature of the constructor of all existing record payload implementations intended to be used by the user has to be changed.
- For payload classes that are not affected by the bug, keep both old (without props) and new constructors with props for backwards compatibility. For payload classes affected by the bug, i.e., extending DefaultHoodieRecordPayload, only the constructors with props are kept. We change all payload classes so that the new constructor is called once in reflection, instead of twice with the first attempt failing if not including new constructors with the properties, to avoid unnecessary fallbacks and performance hits on each record.

  @Deprecated
  public OverwriteWithLatestAvroPayload(GenericRecord record, Comparable orderingVal) {
    this(record, orderingVal, EMPTY_PROPS);
  }

  @Deprecated
  public OverwriteWithLatestAvroPayload(Option<GenericRecord> record) {
    this(record, EMPTY_PROPS);
  }

  public OverwriteWithLatestAvroPayload(GenericRecord record, Comparable orderingVal, Properties props) {
    super(record, orderingVal, props);
  }

  public OverwriteWithLatestAvroPayload(Option<GenericRecord> record, Properties props) {
    this(record.isPresent() ? record.get() : null, 0, props); // natural order
  }

Moves general payload creation methods from DataSourceUtils to HoodieRecordUtils and modified them according to the constructor changes.
Fixes all places around instantiating HoodieRecordPayload to properly pass the configurations in Properties.
To fix the logic for determining deletes, BaseAvroPayload#isDeleteRecord now takes the configuration in Properties, which is called by the new constructors of the payload classes. The old method is still kept for backwards compatibility.

protected boolean isDeleteRecord(GenericRecord genericRecord, Properties props)

DefaultHoodieRecordPayload overrides the implementation of isDeleteRecord to use custom delete key and marker to determine deletes, which is the behavior before RFC-46.
Adds new tests to validate that DefaultHoodieRecordPayload and deletes with custom delete key and marker work as expected.

Impact

Fixes the DefaultHoodieRecordPayload to properly handle custom delete key (hoodie.payload.delete.field) and delete marker (hoodie.payload.delete.marker. Deletes using default _hoodie_is_deleted field are not affected, i.e., working before this fix.

Risk level

medium

Documentation Update

HUDI-6966 for updating docs on custom payload implementation.

Contributor's checklist

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

…rdPayload

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

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

hudi-bot · 2023-10-23T00:51:33Z

CI report:

355a341 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

yihua · 2023-10-23T21:58:07Z

@danny0405 I also changed the payload creation logic for Flink. Could you also review the relevant changes?

nsivabalan

LGTM. just 1 minor comment

nsivabalan · 2023-10-25T03:10:35Z

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

  private final Comparable<?> orderingVal;

  public HoodieAvroPayload(GenericRecord record, Comparable<?> orderingVal) {
+    this(record, orderingVal, EMPTY_PROPS);


shouldn't we mark these as deprecated ?

Based on my understanding this is only used internally and props are not used, so I don't mark it as deprecated.

danny0405 · 2023-10-25T03:55:29Z

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

      Constructor<?> constructor,
      @Nullable String preCombineField) {
    this.shouldCombine = shouldCombine;
+    this.shouldUsePropsForPayload = shouldUsePropsForPayload;


shouldUsePropsForPayload should be always true?

No, for the record payload class implemented by a user outside Hudi repo, the class may have the constructor with old signature, i.e., {GenericRecord.class, Comparable.class} or {Option.class}, so to be compatible on that, there's a fallback mechanism to create payload without properties (which is OK is the payload does not leverage any props). See PayloadCreation#instance.

danny0405 · 2023-10-25T03:57:13Z

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

+  }
+
+  public static Properties extractPropsFromConfiguration(Configuration config) {
+    Properties props = new Properties();


If all we want is payload properties, you can use StreamerUtil.getPayloadConfig.

We need hoodie.payload.delete.field and hoodie.payload.delete.marker here, which are not included in StreamerUtil.getPayloadConfig

Just set it up correctly in the code. BTW, Flink never supports these 2 options I think.

danny0405 · 2023-10-25T04:23:39Z

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

-    super(record, orderingVal);
+  public DefaultHoodieRecordPayload(GenericRecord record, Comparable orderingVal, Properties props) {
+    super(record, orderingVal, props);
  }


The source of the props seems a chaos, I already saw several ways how it was produced:

config.getPayloadConfig().getProps() in HoodieMergeHandle;

payloadProps.setProperty(HoodiePayloadProps.PAYLOAD_ORDERING_FIELD_PROP_KEY, preCombineField); in HoodieFileSliceReader;

config.getProps() in HoodieIndexUtils.

Yeah, this should be cleaned up separately. Regardless, the payload-related configs should always be included so there's no correctness issue. And we're not storing any props with the record so record processing is consistent.

danny0405 · 2023-10-25T04:24:28Z

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

+    final String deleteKey = props.getProperty(DELETE_KEY);
    if (StringUtils.isNullOrEmpty(deleteKey)) {
-      return isDeleteRecord(genericRecord);
+      return super.isDeleteRecord(record, props);


Is this line the actual fix, I didn't see the props got used by the super method, so do we still need to pass around all the props here?

The implementation of this method is kept the same. However, before my fix, this method isDeleteRecord(GenericRecord genericRecord, Properties properties) was a protected method not overriding any method in super class, and it was not called in the constructor to determine whether a record is a delete.

After the fix, the method isDeleteRecord(GenericRecord genericRecord, Properties properties) overrides the same in the super class, and is now called in the constructor (see super class BaseAvroPayload constructor, where the following is called)

this.isDeletedRecord = record == null || isDeleteRecord(record, props);

But the BaseAvroPayload.isDeleteRecord does not really use the passed in props, so do we still to change all the constructors just for passing the props, it seems just an additional invocation of BaseAvroPayload.isDeleteRecord could fix the problem and this method does not need any props actually.

We can address the props issue is separate task maybe.

The BaseAvroPayload.isDeleted(Schema schema, Properties props) already passes around the properties, so can we just overwrite it for DefaultHoodieRecordPayload and there is no need to change all the constructors?

BaseAvroPayload.isDeleteRecord(GenericRecord genericRecord, Properties props) uses the props, and it is called by this.isDeletedRecord = record == null || isDeleteRecord(record, props); in the constructor. BaseAvroPayload.isDeleteRecord(GenericRecord genericRecord) is only for backwards compatibility, called from user-implemented record payload implementation.

I'm talking about the maste code, in master, BaseAvroPayload.isDeleteRecord only handles _hoodie_is_deleted field which looks reasonable. If BaseAvroPayload.public boolean isDeleted(Schema schema, Properties props) is the culprit that incurs the issue then just fix it, we can override it in DefaultHoodieRecordPayload and utilize those specific options there.

The intention is that isDeleted(Schema schema, Properties props) should not deserialize the bytes for perf reason so we cannot implement the custom delete marker there for DefaultHoodieRecordPayload. The way now in this fix is to check the delete key and marker in the constructor before the Avro record is serialized; that’s why props needs to be passed in to the constructor.

nsivabalan · 2023-11-15T19:56:27Z

hey @yihua @danny0405 : can you folks sync up and resolve soon. We wanted to get this landed for 0.14.1.

nsivabalan · 2023-11-21T10:01:19Z

attempting a diff approach here #10150

[HUDI-6961] Fix deletes with custom delete field in DefaultHoodieReco…

d2fd9c4

…rdPayload

nsivabalan added the priority:critical Production degraded; pipelines stalled label Oct 19, 2023

nsivabalan reviewed Oct 19, 2023

View reviewed changes

yihua added the release-0.14.1 label Oct 20, 2023

yihua added 9 commits October 19, 2023 23:24

Fix build after rebasing master

721f4ef

Keep HoodieAvroPayload the same

8a3d6d6

Fix backwards compatibility in BaseAvroPayload

f8cee72

Add a Spark datasource test and fix issues

c580c03

Fix one more reflection issue

db35ccd

Add more code for backwards compatibility

e73788c

Fix compatibility on old constructors

187aa0b

Add verification on deletes in TestDefaultHoodieRecordPayload

f2a25c4

Fix HoodieAvroPayload

088db01

yihua added the writer-core label Oct 21, 2023

yihua added 2 commits October 21, 2023 15:24

Fix nits

d6eab69

Fix test

355a341

yihua marked this pull request as ready for review October 21, 2023 22:50

apache deleted a comment from hudi-bot Oct 22, 2023

yihua requested a review from danny0405 October 23, 2023 21:58

nsivabalan approved these changes Oct 25, 2023

View reviewed changes

danny0405 requested changes Oct 25, 2023

View reviewed changes

danny0405 reviewed Oct 25, 2023

View reviewed changes

nsivabalan self-assigned this Nov 15, 2023

nsivabalan mentioned this pull request Nov 21, 2023

[HUDI-6961] Fixing DefaultHoodieRecordPayload to honor deletion based on meta field as well as custome delete marker #10150

Merged

4 tasks

nsivabalan closed this Nov 24, 2023

[HUDI-6961] Fix deletes with custom delete field in DefaultHoodieRecordPayload #9892

[HUDI-6961] Fix deletes with custom delete field in DefaultHoodieRecordPayload #9892

Uh oh!

Conversation

yihua commented Oct 19, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Change Logs

Impact

Risk level

Documentation Update

Contributor's checklist

Uh oh!

Uh oh!

Uh oh!

Uh oh!

hudi-bot commented Oct 23, 2023

CI report:

Uh oh!

yihua commented Oct 23, 2023

Uh oh!

nsivabalan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yihua Oct 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danny0405 Oct 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danny0405 Oct 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

danny0405 Oct 25, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nsivabalan commented Nov 15, 2023

Uh oh!

nsivabalan commented Nov 21, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

yihua commented Oct 19, 2023 •

edited

Loading

yihua Oct 25, 2023 •

edited

Loading

danny0405 Oct 25, 2023 •

edited

Loading

danny0405 Oct 25, 2023 •

edited

Loading

danny0405 Oct 25, 2023 •

edited

Loading