apache / apache/hudi

[SUPPORT] Problem when refactor a custom payload to new API defined in RFC-46

Open
#9,430 11 comments 0 reactions 0 assignees View on GitHub
issue:function-regression
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.