Flink - Demonstrate (position-based) bug with multiple upserts to same row

[kbendick]

As a follow up to #4364, it was discovered that there is an existing bug that is likely related to the positional deletes generated during a single snapshot.

Opening this draft so that others can inspect it. Will also be opening an issue for it as well.

[kbendick]

cc @openinx @stevenzwu @hililiwei you might be interested in this additional bug I found when using Flink in upsert mode.

Of these two tests, the first one does not pass, and the second one does (as noted).

[kbendick]

This fails every time, with the actual row for ("aaa", '2022-03-01') being Row.of("aaa", dt, 6, true)

[hililiwei]

In my local environment, it seems to be random and occasionally successful. I'm digging deeper to find out why.

[hililiwei]

cc @openinx @stevenzwu @hililiwei you might be interested in this additional bug I found when using Flink in upsert mode.

Of these two tests, the first one does not pass, and the second one does (as noted).

Looks interesting. Let me try. Thanks for Ping me. 😃

[kbendick]

This works correctly.

[kbendick]

The usage of TO_DATE causes calcite to generate a plan which adds a union across all of the VALUES entries, causing them to potentially be omitted in parallel which makes the plan not what is expected.

This is discussed in length here: #4515

[rdblue]

I just spent some time debugging this and I think the problem is with Flink, not with Iceberg.

When I run the case that fails, the issue is that the upserted rows are out of order. In both the final insert data file and the records that I see in the debugger, the column with bool=true is passed in last. Because this is an upsert, the last row is the version that you end up with. I'm not sure why Flink is passing the rows in the wrong order, but it could be that the parser doesn't keep track of the order of rows in a VALUES clause.

I also did a little exploration into what is different with the "succeeds" case. The significant change between the two in my testing was the use of TO_DATE('2022-03-01') instead of DATE '2022-03-01'. TO_DATE appears to trigger the rows getting out of order.

I'm going to close this issue. Feel free to reopen if you think it is a problem with Iceberg somehow reordering the rows, but that seems unlikely given that these rows are coming from Flink and are identical when they come through the Iceberg API (the TO_DATE / DATE distinction is internal to Flink).