[SUPPORT] Global index upsert when using flink and FLINK_STATE index type.
- 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.