apache / apache/hudi

[SUPPORT] Global index upsert when using flink and FLINK_STATE index type.

Open
#8,203 2 comments 0 reactions 1 assignee Claimed by @danny0405 View on GitHub
area:streaming engine:flink priority:high type:feature
Dominant language
Java
Stars
6.2k
Forks
2.5k
Avg merge
2d 8h
Merged PRs (30d)
111

Description

**_Tips before filing an issue_**

Hi, team!

Now when partitioning upsert, according to `precombine.field` parameter, keep the record with the largest value after upserting.

This is widely used to solve the case of out-of-order data, by setting the `precombine.field` to the event time to keep records with the largest event time.

However, when using the FLINK_STATE index type, if cross-partition occurs, the precombine.field parameter will not fully take effect.

In the case of cross-partitioning, the master's logic uses data that arrives later, even if the event time is smaller.

https://github.com/apache/hudi/blob/b79ce80f709ddffb5ad7aead10711222c086db24/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java#L181-L184

I'm wondering if there is any particular reason for implementing it this way?
If not, I'll open a PR to fix it.

It may be necessary to unify the logic of intra-partition upsert and cross-partition upsert, which is convenient for users to understand and use.

**Environment Description**

* Hudi version : 0.13.0

* Hive version : 2.3.9

* Hadoop version : 2.8.5

* Storage (HDFS/S3/GCS..) : OSS

* Running on Docker? (yes/no) : no

Contributor guide

No contributing guide indexed for this repository

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.