[HUDI-115] Enhance OverwriteWithLatestAvroPayload to also respect ordering value of record in storage

[bhasudha]

Tips

• Thank you very much for contributing to Apache Hudi.
• Please review https://hudi.apache.org/contributing.html before opening a pull request.

What is the purpose of the pull request

This enhances the default payload class to respect records in disk when combining. If the records in disk is more recent they are rewritten with current commit time.

(For example: This pull request adds quick-start document.)

Brief change log

(for example:)

• Modify AnnotationLocation checkstyle rule in checkstyle.xml

Verify this pull request

(Please pick either of the following options)

• This pull request is a trivial rework / code cleanup without any test coverage.

(or)

• This pull request is already covered by existing tests, such as (please describe tests).

(or)

• This change added tests and can be verified as follows:

(example:)

• Added end-end test to TestDataSource to test writes from datasource
• Added unit tests to TestDataSourceDefaults

Committer checklist

• Has a corresponding JIRA in PR title & commit

• Commit message is descriptive of the change

• CI is green

• Necessary doc changes done or have another open PR

• For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.

[n3nash]

Can you describe any more use-cases apart from the ordering field for the Map<> introduced ? Why can't the user simply perform this during the old combineAndGetUpdateValue by comparing the incoming ordering field val vs the old one (in the GenericRecord), what am I missing ?

[bhasudha]

Map was a more generic structure to send properties. For now, its only the ordering field that is required here. For your second question, not sure if I am understanding it right. Are you suggesting, we send in the ordering field val itself instead of the field name ?

[n3nash]

Let me explain. The current interface combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema) doesn't allow the user to pass the ordering field names. You're trying to introduce this and have a Map<> to allow the user to do this. (The description link in the JIRA ticket doesn't work so I needed more context).
Now, since the user can have their their own implementation of combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema), they can have the ordering field name in their own class and maintain it themselves, thus not requiring the need to explicitly pass these names through a method.
I see that the intention of this PR to support this functionality specifically for the OverwriteWithLatestAvroPayload. For preCombine, will we just continue to support natural ordering which may not be the same as for combineAndGetUpdateValue, correct ?

[bhasudha]

Thanks for the clarification @n3nash . I am not able to open the link either. It could be because the initial reporter had different jira ids or something else. But the comments section has good context on what this ticket is about. Also, this has come across multiple times in Slack channels from different users. Based on that I can summarize as follows.

Users are looking to consider records in disk when using OverwriteWithLatestAvro payload class. In preCombine step - we pick the latest value for every key from a batch of input data based on some ordering field. At this step we dont know what is in storage yet. When we are ready to write the records, we iterate the records in storage, for each record determine if there is a matching entry in the input batch. If so, we invoke combineAndGetUpdateValue and pass in the record in disk as the currentValue param. In this step specifically, OverwriteWithLatestAvro can possibly overwrite an already stored record with an older update since we werent comparing the ordering val of new data with data in disk. This whole PR is to provide that capability instead of asking Users to write their own implementation to do this. w.r.t preCombine and combineAndGetUpdateValue both will be using the same ordering field(s) to determine the latest.

In order to not disrupt other payload classes, I deprecated the existing combineAndGetUpdateValue, extended it and provided a default implementation that would ignore the Map field and internally do the old way. OverwriteWithLatestAvroPayload alone will override this default functionality to achieve the above purpose.

Hope that helps!

[n3nash]

@bhasudha thanks for explaining, although my metapoint is still not resolved. If you look at the log scanner -> https://github.com/apache/hudi/blob/master/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieMergedLogRecordScanner.java#L114, the preCombine method is used to combine records written to the log (over multiple batches). Ideally, according to your use-case, since during logging we don't know what is the order of updates, we simply write to disk and leave the resolution for a later time. In this scenario, since combineAndGetUpdateValue is not called, how will preCombine honor the orderingVal ?

[bhasudha]

@n3nash preCombine is using the same orderingVal as would combineAndGetUpdateValue. And in the HoodieMergedLogRedordScanner class we are trying to resolve records that are logged to the system already and use orderingVal to identify the most recent one. I dont understand why would we need to use combineAndGetUpdateValue here.

[advancedxy]

Just my two cents:
The interface changes looks a bit of overkill for this type of new feature.

If we are still relying on avro schema, we may leverage schema.getProp(orderingFieldKey) to get ordering field for this record.

[bhasudha]

@advancedxy I guess we would nt know which is the orderingFiledKey when calling this method. That is why we would need to send in that info using another param, so we can get fetch that from the schema later.

[n3nash]

Is this change expected ? Have the styling changed to have the static methods be imported instead of the class ?

[bhasudha]

same as above.

[n3nash]

same static imports question style question, has this changed recently ?

[bhasudha]

nothing changed. Added static import to avoid multiple reference in code.

[bhasudha]

@n3nash could you PTAL when you can ?

[n3nash]

@bhasudha The PR looks good to me. Looks like the same ordering field will be honored in all places. One high level question before I accept it -> If preCombine & combineAndGetUpdateValue are using the same orderingVal, I'm guessing it is expected from the user to use the constructor with the orderingVal and up to the user to ensure the orderingVal used in the constructor is the same as the one passed in Map<..>. If this is true, does HoodieDeltaStreamer allow for this kind of constructor invocation ?
Also, please rebase and push the PR, once the build succeeds can merge it.

[n3nash]

@bhasudha let me know if you need any clarification

[bhasudha]

@bhasudha The PR looks good to me. Looks like the same ordering field will be honored in all places. One high level question before I accept it -> If preCombine & combineAndGetUpdateValue are using the same orderingVal, I'm guessing it is expected from the user to use the constructor with the orderingVal and up to the user to ensure the orderingVal used in the constructor is the same as the one passed in Map<..>. If this is true, does HoodieDeltaStreamer allow for this kind of constructor invocation ?
Also, please rebase and push the PR, once the build succeeds can merge it.

@n3nash Thanks! Yes user needs to ensure that the Map uses the same orderingVal as the one configured for precombine. We are kind of abstracting this logic for users by passing in the precombine field ourselves in the HoodieMergeHandle where this method is invoked. In DeltaStreamer this constructor invocation is happening already here -

hudi/hudi-utilities/src/main/java/org/apache/hudi/utilities/deltastreamer/DeltaSync.java

Line 341 in 0cb24e4

HoodieRecordPayload payload = DataSourceUtils.createPayload(cfg.payloadClassName, gr,

which passes the cfg.sourceOrderingField for ordering val. And in this PR we are also setting the same cfg.sourceOrderingField in the https://github.com/apache/hudi/pull/1704/files#diff-2f9f867c64be16dd87a816fe90f88e87R517 which is what is used in the MergeHandle - https://github.com/apache/hudi/pull/1704/files#diff-78cfbf493ffb0a24c10b52d9602531a1R222 . Hope that answers your question.

[bhasudha]

@xushiyan This PR fails CI due to log limits over threshold - https://travis-ci.org/github/apache/hudi/builds/711685577. I see this error message consistently - The job exceeded the maximum log length, and has been terminated. . Wondering if you have any idea on this ?

[xushiyan]

Hi @bhasudha this means there are too many logs got printed out during the test. In ideal scenario, a test does not output anything unless there is a failure. Could you check if any log or print statement can be removed?

[bhasudha]

created a Jira issue to track this separately - https://issues.apache.org/jira/browse/HUDI-1127

[nsivabalan]

@bhasudha : I haven't reviewed the PR, but a high level question. do we have any guards if user changes the ordering field prop along the way ? Basically I expect we fail any operation if ordering field prop is changed.

[nsivabalan]

if schema got evolved and the ordering field for incoming record does not exist in existing storage, this might throw NPE.

[bhasudha]

@bhasudha : I haven't reviewed the PR, but a high level question. do we have any guards if user changes the ordering field prop along the way ? Basically I expect we fail any operation if ordering field prop is changed.

This will be handled the same way our ingestion works when ordering field changes. in the merge phase, the new ordering field will be used to compare the incoming and persisted record.

[bhasudha]

if schema got evolved and the ordering field for incoming record does not exist in existing storage, this might throw NPE.

Thanks @nsivabalan nice catch. I think if the incoming records schema evolved to add new columns and they are also used for ordering, the persisted record will not have that ordering field. And that will throw an NPE. I ll handle that here.

[nsivabalan]

leaving one comment.

[n3nash]

@bhasudha let me know once you've addressed @nsivabalan comment and the build is passing, can merge this.

[shenh062326]

I am working on https://issues.apache.org/jira/browse/HUDI-1058 , it also needs similar modifications. It is better to work on HUDI-1058 after this merge request is merged, but if this MR has not progressed, I can resolve HUDI-1058 first.

[vinothchandar]

Looks like this is almost ready . I will rebase, review and try to land this first.

thanks for flagging @shenh062326 !

[vinothchandar]

@bhasudha as discussed, should we try and have another payload for this behavior. to avoid overwriting for backfill scenarios

[bhasudha]

@bhasudha as discussed, should we try and have another payload for this behavior. to avoid overwriting for backfill scenarios

yeah makes sense. I can put up a PR with another payload @vinothchandar

[bhasudha]

closing this for now.

[FelixKJose]

@bhasudha What is the new PR ( as you mentioned "I can put up a PR with another payload @vinothchandar"). Or this feature is implemented already? Which PR is used to track late arrived latest records?

[vinothchandar]

@FelixKJose I don't think there is one up yet. We can get to this in the next few weeks, if you are in need of this.

[FelixKJose]

@vinothchandar Thank you for the reply.

Yes, this will be very important feature for us, since we are relying on async messaging as source, so out of order message is a real scenario and this feature will handle those out of order message scenario.

[advancedxy]

@FelixKJose I don't think there is one up yet. We can get to this in the next few weeks, if you are in need of this.

I think this is also a big issue for my cases here. @bhasudha or @vinothchandar is it possible the new payload will be ready in the next week?

I can contribute the new implementation if you are lacking of time.

[bhasudha]

@FelixKJose I don't think there is one up yet. We can get to this in the next few weeks, if you are in need of this.

I think this is also a big issue for my cases here. @bhasudha or @vinothchandar is it possible the new payload will be ready in the next week?

I can contribute the new implementation if you are lacking of time.

@advancedxy That would be very useful. I can help you if you have any questions. Please feel free to send a PR with the new payload.

[nsivabalan]

@vinothchandar : may I know whats the consensus here. Let me take a stab, but do confirm the same.
Introduce a new class for payload class that supports this specific ordering. This payload class will have the following api

Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema, Map<String, String> props) throws IOException;

Will this extend from existing HoodieRecordPayload? bcoz, this the interface used everywhere right?

Or did you intend a class as follows

class HoodieRecordOrderingPayload implements HoodieRecordPayload{
 String orderingField; // value passed in via constructor. 
 
 Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema) throws IOException{
    // honor ordering value and return
 }
.
.
}

[nsivabalan]

@vinothchandar @bhasudha : can you throw some light on why new config rather than reusing preCombine field value? I am working on putting up a new PR, so would like to know was this intentional or unintentional.

[vinothchandar]

That's because you don't know what the ordering field value is, when reading the data from storage right?

[nsivabalan]

ok, so for preCombine (combining two records within the same ingestion batch) we use one config and for determining ordering among one incoming record and another in storage or in other words, to combine an incoming record w/ one in storage, we use a diff config?
I see both these as one and the same. just that whether we resolve records within same batch or across batches is what differs.

[vinothchandar]

You are right, they are one and the same. If we can avoid the additional config, we should

[nsivabalan]

gotcha. here is my take.
Will introduce a config called "honorOrderingToCombineRecordsAcrossBatches" (we can work out a good naming). default value is false. Will send in orderingFieldKey (same as preCombineFieldKey) as an arg to OverwriteWithLatestAvroPayload.

public OverwriteWithLatestAvroPayload(GenericRecord record, Comparable orderingVal, 
String orderingFieldKey, boolean honorOrderingToCombineRecordsAcrossBatches) {
.
.

Based on "honorOrderingToCombineRecordsAcrossBatches" value, combineAndGetUpdateValue() impl will decide to go with existing impl or new one (as we see in this patch)

I know we do have a gap here where in, if ordering field has changed over time, we can't do much here. But we can't afford to store the ordering field as a separate column in dataset either, as diff commits could theoretically have diff ordering field. But guess we can call it out that we may not support such evolution of ordering field.

[vinothchandar]

I prefer we don't do this. but separate the existing impl and new one into separate payload classes.

don't understand actually how we plan to implement obtaining the ordering field from the record in dfs. could you please clarify? @nsivabalan

[nsivabalan]

similar to how this patch already does. basically we lookup the preCombine field in existing generic record.

Object persistedOrderingVal = getNestedFieldVal((GenericRecord) currentValue, props.get(ORDERING_FIELD_OPT_KEY), true);

Reason I proposed to tweak existing class is.
Even if we want to go w/ 2 classes, Ideally I would want to name existing class as OverwriteWithIncomingPayload and introduce a new class called OverwriteWithLatestAvroPayload, bcoz, existing one overwrites w/ incoming and we are introduce a new class which will actually overwrite w/ latest payload.
But since users configure via class names, we can't go w/ this approach and hence my proposal to introduce a config and change existing class.