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

[BruceKellan]

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

[danny0405]

It's kind of a trade-off, we can implemet the coss-partition update only if we record the version field for each index item, that can increase the index storage significantly, but I think it is feasible.

[BruceKellan]

Thanks for your reply, danny. I will open a ticket to do it.