Fix for table upsert mode for Flink 1.12 actually works, the problem is flink itself

[kingwind94]

Apache Iceberg version

0.13.1

Query engine

Flink

Please describe the bug 🐞

According to #4519, the fix for the issue did not resolve the issue in Flink 1.12: #4418. So an error message about correctness issue is logged and nothing fixed.

But acutually, the TestFlinkUpsert test cases not passed is actually because in flink 1.12 the "Insert into table values ... " has a correctness issue for parallelism greater than 1. There is a NotNullEnforcer operator between the insert values source and iceberg sink, and the operator is with parallelism equal to the job parallelism, what is more, the data is transfer in REBALANCE strategy from the source to NotNullEnforcer operator, which corrupts the data order.

So if you change the job parallelism to 1, the test cases should be all passed.

[github-actions]

This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.

[github-actions]

This issue has been closed because it has not received any activity in the last 14 days since being marked as 'stale'