Flink: In upsert mode with equality fields on a partitioned table, partition column has to be in equalityFields

[tomerr90]

Apache Iceberg version

1.10.1 (latest release)

Query engine

Flink

Please describe the bug 🐞

1. https://github.com/apache/iceberg/blob/main/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L952

2. https://github.com/apache/iceberg/blob/main/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java#L755

This validation is wrong... it seems to assume in the case above (upsert + partition + equalityFields) distribution mode has to be 'hash', which it doesnt, if its left at 'none', we'll have rows with the same equalityFields routed to the same writer to handle deletes properly, and a potentially a number of writers writing to the same partition, which is completely fine.
It is also just plain wrong sice if I have something like create time as a partition field, which is pretty common, equality will fail for the same row id in different times, maybe the intention was to make the partition transform value part of the equality fields? That would maintain correctness, and also help with writer explosion
Am I misunderstanding something?

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[sungwy]

Hi @Tomer90 - thanks for raising this issue.

I took a look at the second code reference you shared and I suspect one of the following is true about the invariant for partition source IDs to be a part of equality fields for upserts:

1. The invariant is only needed for hash distribution mode. Then the implementation is wrong and should gate this check on the actual distribution mode like you suggest.
2. The invariant is correct independent of distribution mode. Then the error message is simply wrong and should be updated.

We may benefit from pulling in Flink Iceberg experts here, but based on the Javadocs I think the second conclusion is likely to be true.

Let me know if the explanation in the javadocs aligns with your understanding of how upserts should work as well

[tomerr90]

Thanks Sung!
I feel like I have to be missing something here, an Iceberg table partitioned by a function over a timestamp field is probably the most common use case of Iceberg, so I find it very weird that this common case basically doesnt work in Flink upsert because it forces you to use the timestamp field (without the function even, just the field) as an equality field for the delete.

@mxm @rodmeneses @Guosmilesmile sorry for the tag, PLMK if thats not OK, I saw you guys worked in the IcebergSink recently, would greatly appreciate your take on the above, seems like this "double" use of equalityFields for deletion equality and also flink stream partitioning is limiting common use cases, maybe the keyBy in the sink should transparently add the partitioning functions of the table AND the equalityFields? So equality is still only checked by the equalityFields but we maintain correctness of the deletes.
Also, in general, this whole discussion is only relevant if writeParallelism is > 1, agree?

[tomerr90]

Apologies, just thought about it again and realised Im wrong, the partition key needs to be consistent and not changing and create_time is like that so I can add it as an equality field, even though its a bit inefficient, but it will still work

[mxm]

Just for context: We have to make sure a given equality key value gets processed by the only one writer. Otherwise, multiple writers would be writing delete files which could conflict. For non-partitioned table, we partition by the equality value itself. For partitioned table, like in your case, we enforce that the equality values are a subset of the partition key.