[HUDI-2815] add partial overwrite payload to support partial overwrit…

[stayrascal]

…e case

Tips

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

What is the purpose of the pull request

Add a customized payload PartialOverwriteWithLatestAvroPayload to support partial overwrite records fields.
For example, two incoming record batch A(1, null, 1, 1) and B(2, 2, null, 2) will be combined as the result (2, 2, 1, 2)
(For example: This pull request adds quick-start document.)

Brief change log

• Add a PartialOverwriteWithLatestAvroPayload to support partial overwrite records
• Add a config partial.overwrite.enabled to control if enable partial overwrite
• Change the ValueState of BucketAssignFunction from HoodieRecordGlobalLocation to HoodieRecord
• Support partition path changed case
  • Merge HoodieRecord before storing the record in ValueSate
  • Copy and sink a new record from state to downstream operator while the record partition is changed
• Add compareTo to compare two HoodieRecords to chose their sequence, because cannot get the sequence from #preCombine method if merge two records in #preCombine method
• Add schema filed(String) in new AvroPayload, which will be used to #preCombine two records
• Change BootstrapOperator to load HoodieRecord with data instead of only HoddieKey
• Change RowDataToHoodieFunction to create payload with schema

Verify this pull request

• Added PartialOverwriteWithLatestAvroPayloadTest to verify the change.
• Manually verified the change by running a job locally with both preCombine and combineAndGetUpdateValue methods

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.

[stayrascal]

@hudi-bot run azure

[stayrascal]

@hudi-bot run azure

[danny0405]

Thanks for the contribution ~ I kind of think we better do this after id-based schema evolution is supported, only after that, we have more light-wright solution to support per-record schema.

Generally, take schema with record seems not a good solution.

[danny0405]

Why we must need a compareTo here ?

[stayrascal]

The previous logic of data2.preCombine(data1) is that return one of data1 or data2 ordering by their orderVal. But if we merge/combine data1 and data2 into a new payload(reduceData), the data1.equals(reduceData) is always false. In order to get the HoodieKey and HoodieOperation for new HoodieRecord with reduceData, we need to get the latest HoodieKey and HoodieOperation from data1 and data2， compareTo is used for replace #preCombine to compare their orderingVal.

 @Override
  public int compareTo(OverwriteWithLatestAvroPayload oldValue) {
    return orderingVal.compareTo(oldValue.orderingVal);
  }

@Test
  public void testCompareFunction() {
    GenericRecord record = new GenericData.Record(schema);
    record.put("id", "1");
    record.put("partition", "partition1");
    record.put("ts", 0L);
    record.put("_hoodie_is_deleted", false);
    record.put("city", "NY0");
    record.put("child", Arrays.asList("A"));

    PartialOverwriteWithLatestAvroPayload payload1 = new PartialOverwriteWithLatestAvroPayload(record, 1);
    PartialOverwriteWithLatestAvroPayload payload2 = new PartialOverwriteWithLatestAvroPayload(record, 2);

    assertEquals(payload1.compareTo(payload2), -1);
    assertEquals(payload2.compareTo(payload1), 1);
    assertEquals(payload1.compareTo(payload1), 0);
  }

[stayrascal]

Actually, rec1 and rec2 should have same HoodieKey here, right, but the HodieOperation might different.

[danny0405]

We should be caution of the DELETEs, should we still merge for DELETE messages ?

[stayrascal]

yeah, if one of record is DELETE record, just return themselves directly, no need to merge, the DELETE message to delete old record during hudi write. Only when two records are not DELETE records, we need to merge them.

[danny0405]

This would increase the state size significantly. We better avoid this with better solution, and why we must sore full record instead of index here ?

[stayrascal]

Thanks @danny0405 for reviewing this.

Yeah, as mentioned in this issue #4030.
The reason why we need full record here is want to handle the case that the record partition path is changed.

In the original logic, once the record partition path changed, will sink a delete record to old partition file to delete old record, and then sink the incoming record with new partition to new partition file, the final record will only contains the info from incoming record and miss the info from old record. (note the OverwriteNonDefaultsWithLatestAvroPayload also have this issue)

So we need to retrieve the old/existing record from base file, and then merge/combine with incoming record, as currently we don't support lookup record from base file, so we have to store the old/existing in somewhere. e.g. fink state. BucketAssignFunction is the only place we can store the old/existing record and change its location from old partition file to new partition file.

So the new logic is that:

• store the old record(from source or base file by enable Bootstrap) in Flink state
• once a new record coming with same record key but the partition changed
  • sink a delete record to old partition file to delete file
  • retrieve & copy old record from state and change its location with new partition, and sink to new partition file
  • the copied record and incoming record will be merged by #preCombine

the drawback here is that it will increase the state size, but if we don't use the state to store full record, it seems that we don't have approach to merge incoming record with existing record in base file while partition change.

I also consider this problem, what I'm thinking to avoid impact the current logic(overwrite with latest payload) is that create a updateState abstract method and treat indexState as a abstract filed, different sub class will implement the logic with ValueState<HoodieRecordGlobalLocation> or ValueState<HoodieRecord>, or a StateHelper to handle state operation.

[stayrascal]

Thanks for the contribution ~ I kind of think we better do this after id-based schema evolution is supported, only after that, we have more light-wright solution to support per-record schema.

Generally, take schema with record seems not a good solution.

Yeah, agree with taking schema with record is not a good solution, it couple the byte data with schema and hard to evolute schema later, but it seems that we cannot merge two Payload without schema. I also tired if we can pass the schema to #preCombine method, just similar with #combineAndGetUpdateValue. But we only can get the schema info from flink configuration, the FlinkWriteHelper doesn't(and should not) support flink configuration to support retrieve schema from configuration, so I have to inject schema via constructor.

[nsivabalan]

CC @xushiyan @alexeykudinkin : who might be working on refactoring payload interfaces. and @xiarixiaoyao who might be working on schema evolution story. can you folks take a look at the patch.

[stayrascal]

CC @xushiyan @alexeykudinkin : who might be working on refactoring payload interfaces. and @xiarixiaoyao who might be working on schema evolution story. can you folks take a look at the patch.

Thanks @nsivabalan to follow up. Take a summary, the current implementation has two drawbacks:

• embedding schema in payload might hard to do schema evolution later.
• using flink state to store full HoodieRecord instead of location, will increase the state size.

And there are two challenges need to be solved for supporting partial update.

• how combine/merge two payload in #preCombine if without schema.
• how to lookup the record from base/log file during the record partition patch changed, and then combine/merge with the incoming record later.

Any thoughts on these two challenges?

[nsivabalan]

Left some high level comments. Will let Danny follow up on reviews since this fixes flink code base.

[nsivabalan]

this might be confusing w/ schema arg with combineAndGetUpdateValue. can you fix either of the names.

[nsivabalan]

but in general, storing schema along w/ payload might have an impact on the performance. and thats why initial payload was designed that way. So, do add a line here for payload implementations setting this schema field might have to watch out for performance.

[stayrascal]

removed this field.

[nsivabalan]

guess, this has to be "this.schema.getFields". as I commented earlier, its confusing :) . can we fix the naming of either of them.

[stayrascal]

fixed the schema name.

[nsivabalan]

do we need to deal w/ nested fields here ?

[stayrascal]

current logic will overwrite whole nested field, if the incoming field is not null.

And I think we don't need to support the partial update inner nested field, for example, for Map, List, etc. we should not merge map(1 -> 'a', 2 -> 'b') & map(1 -> ''', 3 -> 'c') to map(1 -> '', 2 -> 'b', 3 -> 'c') incase the upstream want to delete the key '2', if we merge them together, they cannot delete some elements. the List as well.

[nsivabalan]

instead of storing the schema with payload, did you think about adding a new preCombine method as follows

OverwriteWithLatestAvroPayload preCombine(OverwriteWithLatestAvroPayload oldValue, Schema schema);

this would make it a lot simpler right. Since preCombine is used only to dedup records within a single batch, both records should have same schema.

[stayrascal]

Hi @nsivabalan , thanks a lot for review this.

Regarding adding new preCombine method with Schema, I considered this, but it means that the method caller who need to get the schema info at first, and currently, it seems that we only can get the schema info from Configuration(from hoodie.avro.schema field). Sometimes, the caller might hard to get the schema info, Especially for FlinkWriteHeler.deduplicateRecords(List<HoodieRecord<T>> records, HoodieIndex<?, ?> index, int parallelism).

But compare the performance, it seems that passing the schema in method might be a better approach.
BTW, since we already have had the method preCombine(T oldValue, Properties properties), how about put the schema string in properties, and then parse the schema string to Schema later, so that we don't need to create a new method any more. otherwise, I cannot image when will we will Properties.

[nsivabalan]

insertRecordOption -> oldRecordOption

[stayrascal]

solved.

[nsivabalan]

argh. this again clashes w/ instance variable "schema". Can we fix the naming.

[stayrascal]

solved

[stayrascal]

Hi @danny0405 , regarding the two changes:

• embedding schema in payload might hard to do schema evolution later.
• using flink state to store full HoodieRecord instead of location, will increase the state size.

for first one, I remove the schema filed from payload class, instead, by passing the schema to preCombine method, it should be easy to do schema evolution if need.
for second one, I revert my changes about store whole HoodieRecord back to store location, it still has the problem that once the record partition path changed, but i think it should enough for supporting the case that the table don't have partition, or the partition path of record won't changed.

BTW, do you think is it worthy to implement a new bucket assign function(can be controlled by feature toggle/configuration ) which store the full record to support partial update totally ?

[alexeykudinkin]

I think currently established semantic for preCombine -- you select either A or B, but you don't produce new record based on those 2, since it's mostly used to de-dupe records in the incoming batch. I can hardly imagine the case to fuse 2 incoming records into something third. Can you help me understand what use-case you have in mind here?

[stayrascal]

Thanks @alexeykudinkin for reviewing this.

What we are trying to do is implement partial update purpose. For example, let's assume the record schema is (f0 int , f1 int, f2 int), The first record value is: (1, 2, 3), the second record value is: (4, 5, null) with the field f2 value as null. We hope that the result after run preCombine is (4, 5, 3), which means we need to combine/merge two records to a third one, not only choose one of them.

Actually, what we want to implement is similar with #combineAndGetUpdateValue(IndexedRecord currentValue, Schema schema) which used for combine the incoming record with existing record from base/log file.
But #preCombine will be used for combing/merging two incoming records in a batch.

[alexeykudinkin]

Right, that's exactly my question: why do you want to implement such semantic w/in preCombine? What use-case you're trying to accommodate for here?

Essentially with this change you will introduce a way for 2 records w/in the batch to be combined into 1. But why do you need this?

After all you can achieve the same goal if you just stop de-duping your records, and then subsequently merge them against what is on disk

[stayrascal]

Hi @alexeykudinkin, I got your point. if we have to combine two records to a combined one, we'd better to implement the combine logics in other place, maybe in some util or helper classes, or skip the de-duping logic, right?

Here are some options from mine that #preCombine might be a better place to implement these logics, or create new merge method in HoodieRecordPayload interface.

• First, from the description of preCombine method, it used for combining multiple records with same HoodieKey before attempting to insert/upsert to disk. The "combine multiple records" might not mean only choosing one of them, we also can combine & merged them to a new one, just depends on how the sub-class implement the preCombine logic(Please correct me if my understanding is wrong :) ). Yeah, it might be a little bit confused that we need Schema if we are trying to merged them.
• Second, I checked when will we call preCombine method is trying to duplicate records with same HoodieKey before insert/update to disk, especially in Flink write case, even through the duplicated logic is choose the latest record, but we need to ensure that one HoodieKey should only contains one record before comparing to existing record and write to disk, otherwise, some records will missed. For example, in HoodieMergeHandle.init(fieId, newRecordsIter), it will convert the record iterator to a map and treat the recordKey as key. So we might not stop de-duping logics and merge them against what is on disk unless we change the logic here. And also we implement another class/method to handle the merge logic, and switch the existing de-duping logic from calling preCombine to new class/method, we have to add an condition to control whether should we call preCombine or not, I think it might not a good way. Instead, we should handle it in preCombine method by different implemented payload.

That's what my thought here, and I'm glad to listen your useful suggestions. :)

[alexeykudinkin]

Let me try to clarify a few things:

preCombine has a very specific semantic: it's de-duplicating by the way of picking "most recent" among records in the batch. Expectation always is that it being handed 2 records it will have to return either of them. It could not produce new record. If we want to revisit this semantic this is a far larger change that will surely require writing an RFC and broader discussion regarding the merits of such migration. Please also keep in mind that as of RFC-46 there's an effort underway to abstract whole "record combination/merging" semantic out of RecordPayload hierarchy into standalone Combination/Merge Engine API.

First, from the description of preCombine method, it used for combining multiple records with same HoodieKey before attempting to insert/upsert to disk. The "combine multiple records" might not mean only choosing one of them, we also can combine & merged them to a new one, just depends on how the sub-class implement the preCombine logic(Please correct me if my understanding is wrong :) ). Yeah, it might be a little bit confused that we need Schema if we are trying to merged them.

Please see my comment regarding preCombine semantic above. I certainly agree with you that the name is confusing, but i've tried to clear that confusion. Let me know if you have more questions about it.

Second, I checked when will we call preCombine method is trying to duplicate records with same HoodieKey before insert/update to disk, especially in Flink write case, even through the duplicated logic is choose the latest record, but we need to ensure that one HoodieKey should only contains one record before comparing to existing record and write to disk, otherwise, some records will missed. For example, in HoodieMergeHandle.init(fieId, newRecordsIter), it will convert the record iterator to a map and treat the recordKey as key. So we might not stop de-duping logics and merge them against what is on disk unless we change the logic here. And also we implement another class/method to handle the merge logic, and switch the existing de-duping logic from calling preCombine to new class/method, we have to add an condition to control whether should we call preCombine or not, I think it might not a good way. Instead, we should handle it in preCombine method by different implemented payload.

You're bringing up a good points, let's dive into them one by one: so currently we have 2 mechanisms

1. preCombine that allows to select "most recent" record among those having the same key w/in the batch
2. combineAndGetUpdateValue that allows to combine previous or "historical" record (on Disk) with the new incoming one (all partial merging semantic is currently implemented in this method)

You rightfully mention some of the invariants are currently that the batch would be de-duped at certain level (b/c we have to maintain PK uniqueness on disk), and so we might need to shift that to accommodate for case that you have. And that's exactly what my question was: if you can elaborate on use-case that you have at hand that you're trying to solve w/ this PR, i would be able to better understand where you're coming from and what's the best path forward for us here.

Questions i'm looking an answers for are basically following:

1. What's nature of your use-case? (domain, record types, frequency, size, etc)
2. Where requirements for partial updates are coming from?

and etc. I'm happy to set some 30min to talk in person regarding this or connect on Slack and discuss it there.

[stayrascal]

Hi @alexeykudinkin , Thanks a lot for you detail clarification.

1. Regarding the design of preCombine, I'm clear now. I'm sorry I don't know the detail of RFC-46, and also I didn't find the link RFC-46 from here, cloud you please share the link?
2. and regarding the requirements for partial updates/overwrite, I saw some same requirements from community. In my case, generally, we want to build a customer profile with multiple attributes, these attributes might come from different systems, one system might only provides some attributes in a event/record, and two systems might the events/records with different attributes, we should not only choose the recent one, we need to merged them before writing to disk. Otherwise, we have to keep all change logs, and then start a new job to dedup & merge these attributes among the change logs. For example, we have 10 attributes a1-a10(all of them are optional), source system A only has the a1-a5, source system B only has a6-a10, what result we expect is that the final record contains a1-a10, not only a1-a5 or a6-a10. And because we might receive two events/records in same time, they might be in a same batch, that's why we want to merge them before combineAndGetUpdateValue .

[stayrascal]

BTW, thanks a lot for you time, will ping you on slack.

[nsivabalan]

CC @rmahindra123 who encountered a necessity to do preCombine but to combine bits and pieces from both records to return a new one. Rajesh: do you wanna go over your use-case may be.

[stayrascal]

@alexeykudinkin , I'm sorry that I still don't find a suitable time to align with online, may i check any thoughts or suggests on this PR?

[alexeykudinkin]

In general it's better to express common functionality in a way that would allow it to be re-used and adopted in other places: here for ex, we can reuse the same routine of combining 2 records into one, across 2 methods if we properly abstract it

[stayrascal]

abstracted the merge method, but still in current class.

[alexeykudinkin]

What's the idea for this additional configuration (beside the record payload class)?

[stayrascal]

This is the feature toggle to control another change about BucketAssignFunction to support the case the record partition path is changed, but I have removed it. so this feature toggle can be removed as well.

[LinMingQiang]

Do I need to modify the preCombine in the HoodieMergedLogRecordScanner.processNextRecord method? What I understand is that when we read a log file, we need to do deduplication and also call preCombine.

[LinMingQiang]

This place needs to be changed to >= , because when we do not set the preCombine field, the first data will always be used instead of the latest

[stayrascal]

Good point here, thx.

[stayrascal]

HoodieMergedLogRecordScanner

As I know, if we also want to achieve partial update purpose, we just need to pass the Schema here.(Assume this pull request has been approved).

[nsivabalan]

Just FYI for all interested folks.
Precombine is not just used to dedup two records within same incoming batch, but also to deduce the winner when we merge records in LogRecordReader.

[stayrascal]

yeah, @LinMingQiang has mentioned this one above.

From my understanding, if we want to enable "partial update" feature by defining customized payload class, it should running "partial update" in these three cases:

• merged the incoming batch records before write to disk
• read records from the log file(read from MOR table)
• read records from log file and compact into base file

So I also update the HoodieMergedLogRecordScanner.processNextRecord () by passing Schema info, if the use case is not using "partial update" with other payload class, it will use the default preCombine() logic to choose "recent" one.

The current situation is that will we treat preCombine return one of two records, or we can merged them to a new record.

Just FYI for all interested folks. Precombine is not just used to dedup two records within same incoming batch, but also to deduce the winner when we merge records in LogRecordReader.

[alvarolemos]

Just added a comment, suggestion a generalization of the proposed approach :)

[alvarolemos]

@stayrascal,

I really liked the idea of having a record payload that does partial merging. However, if I understood it correctly, what's proposed here is to do so in a very specific way: you're favoring the income record field's values, unless they are null (in which case, you would keep the existing one). I'm not saying this is not valuable, but that the idea of doing partial merging is so good that maybe we could have something more generic. I'm going to suggest a few changes in order to accomplish that:

• Make PartialOverwriteWithLatestAvroPayload an abstract class
• Instead of having mergeFunc as a parameter of the mergeRecord method, it could become an abstract method. This would lead to the removal of the overwriteWithNonNullValue method, which makes this implementation specific to your merging logic
• For the original use case (partial merge favoring non-null values), implement the proposed abstract class and implement the mergeFunc method with what you have in overwriteWithNonNullValue: (first, second) -> Objects.isNull(second) ? first : second

It's just an idea, that could make what you proposed useful for many more use cases. Hope this made sense, and thanks for bringing this idea!

[stayrascal]

Hi @alvarolemos , thanks a lot for your useful suggestion. Yeah, I also considered to abstract the merge logic by using an abstract merge method or passing merge function into a generic function, and I choose the later. The reason as follow:

• the preCombine and combineAndGetUpdateUpdate might have different merge/combine logic, only implement one abstract merge function might not enough for both two cases. For example, these two methods in OverwriteWithLatestAvroPayload have different merge/combine logic.
• In current implementation, actually, the mergeRecord is a generic method even through it's a private method currently, but it don't care the detail merge logic and can be changed to protected/public scope if need. Instead, the overwriteWithNonNullValue is merge implementation in current "Payload", which is wrapper of mergeFunc and we can create two wrappers for preCombine and combineAndGetUpdateValue two scenarios if need, which is similar with what you mentioned about implement detail mergeFunc logic in sub class. We can still inherit this class implement detail mergeFunc logic, and pass to mergeRecord method.
• Another reason why i didn't chose creating abstract class currently is that there will only one sub class, we can refactor it if we have many case need to inherit this class, right now, just make it simple as much as possible.

[stayrascal]

@hudi-bot run azure

[hudi-bot]

CI report:

• 20b1ee4 Azure: FAILURE Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

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

[nsivabalan]

@stayrascal : We landed a partial payload support via #4676.
Let us know if we can close this patch or if its possible to enhance the 4676 or if this patch is addressing something different.