[SUPPORT] Problem when refactor a custom payload to new API defined in RFC-46
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
Dear community:
Some of our internal users are very interested in RFC-46, but they have some custom payloads. Therefore, I need to reconstruct these payloads using the new `HoodieRecordMerger` API. During this process, I found that some capabilities of the original payload seem to be missing in the new `HoodieRecordMerger` API.
An example is a custom payload that simulates the logic of a left outer join, which is named as `LeftOuterJoinPayload`. It's similar to the `PartialUpdateAvroPayload`, with the difference that the `LeftOuterJoinPayload` needs to discard rows that do not meet the left outer join condition before being written to storage.
Illustrating the functioning of this payload with an example, let's assume there are two writers working on writing to a HUDI wide table. Writer1 is responsible for reading data from source table1 and writing it to this wide table, while Writer2 handles data from source table2. The payload performs a left outer join between source table 1 and source table 2.
Schema of source table1 is
{name: id, type: string},
{name: ts, type : long},
{name: name, type: string}
Schema of source table2 is
{name: id, type: string},
{name: ts, type : long},
{name: price, type: string}
Schema of target wide table is
{name: id, type: string},
{name: ts, type : long},
{name: name, type: string},
{name: price, type: string}
Current data:
id ts name price
1 1 name_1 null
Insert data:
id ts name price
1 2 null price_1
2 2 null price_2
2 3 null price_3
The record of id 2 after `precombine` is '2, 3, null, proce_3', it should be **dropped** because it does not match left outer join syntax. In the previous implementation of `LeftOuterJoinPayload`, the **dropped** behavior is done in `getInsertValue` method by return `EMPTY` to **skip** writing this record.
However, when I tried to implement the logical of `LeftOuterJoinPayload` using the new `HoodieRecordMerger` API, I found that there is no way to achieve the aforementioned **dropped** logic. This is because, after the merge action of `precombine` is done, there is no callback API available to invoke custom logic to decide whether **dropped** the record or not before writing it to storage. I initially thought that `HoodieRecord#shouldIgnore` could help me, but I realized it's not the solution. This is because `HoodieRecord` is a generic implementation related to the compute engine, not something customized by the user based on their special business requirements.
I wonder if I missed anything, or if it's necessary to introduce a new API to implement this capability.
Contributor guide
No contributing guide indexed for this repository
Research direction
Start with RFC-46 and the HoodieRecordMerger API, then compare the previous LeftOuterJoinPayload behavior in getInsertValue with the limitations described for HoodieRecord#shouldIgnore. Determine how a merger can signal that a precombined record should be discarded before storage; done means the required capability and API shape are agreed and documented.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100