Flink - Fix incorrect row being written for delta files when using upsert mode

[kbendick]

No description provided.

[hililiwei]

Yes, there is no dedicated Upsert type in flink. When the row is transferred for deletion, I think we should prune the row first and retain only the key field. But it's not doing that at the moment.

[hililiwei]

this.structProjection = StructProjection.create(schema, deleteSchema);
   

 public void delete(T row) throws IOException {
      if (!internalPosDelete(structProjection.wrap(asStructLike(row)))) {
        eqDeleteWriter.write(row);
      }
  }

 public void deleteKey(T key) throws IOException {
     if (!internalPosDelete(asStructLike(key))) {
       eqDeleteWriter.write(key);
     }
  }

In our test case, only id province should be paased to deleteKey. It seems that we should use structProjection to trim row.

[hililiwei]

I found that it seemed that the eqDeleteRowSchema passed in was wrong.

[hililiwei]

By passing in the correct eqDeleteRowSchema, our test cases are now running properly. After you approve my PR, let's see if I can pass the CI.

[kbendick]

Closing and reopening for tests.

[kbendick]

This is the present fix to the current issue.

[kbendick]

This fixes the test cases presented, but causes a number of tests to fail. However, all of those tests are upsert or changelog based tests where the upsert happens on the data key, which is the table's partition key (when partitioned) and also the table's only equality field id.

The tests that fail are all testChangeLogOnDataKey and rtestUpsertOnDataKey. And in the test source, the data key is a special key that's part of the table's configuration.

Of note, tests where we upsert on data and some other field still pass.

So we might need to only use the update delete projection when the equality ids are not the same as the partition ids and update the delete schema accordingly. Or we possibly need to update the test harness source, given that it has special handling for this particular key.

I'd also like to add tests for partition evolution, but this should be addressed first.

> Task :iceberg-flink:iceberg-flink-1.14:test

org.apache.iceberg.flink.TestChangeLogTable > testChangeLogOnDataKey[PartitionedTable=true] FAILED
    java.lang.AssertionError: Should have the expected records for the checkpoint#1 expected:<[Record(1, aaa), Record(1, bbb), Record(1, ccc)]> but was:<[Record(1, aaa), Record(2, aaa, 1, false), Record(1, bbb, 1, false), Record(1, ccc)]>
        at org.junit.Assert.fail(Assert.java:89)
        at org.junit.Assert.failNotEquals(Assert.java:835)
        at org.junit.Assert.assertEquals(Assert.java:120)
        at org.apache.iceberg.flink.TestChangeLogTable.testSqlChangeLog(TestChangeLogTable.java:285)
        at org.apache.iceberg.flink.TestChangeLogTable.testChangeLogOnDataKey(TestChangeLogTable.java:167)

org.apache.iceberg.flink.TestChangeLogTable > testChangeLogOnDataKey[PartitionedTable=false] FAILED
    java.lang.AssertionError: Should have the expected records for the checkpoint#1 expected:<[Record(1, aaa), Record(1, bbb), Record(1, ccc)]> but was:<[Record(1, aaa), Record(2, aaa, 3, false), Record(1, bbb, 2, false), Record(1, ccc)]>
        at org.junit.Assert.fail(Assert.java:89)
        at org.junit.Assert.failNotEquals(Assert.java:835)
        at org.junit.Assert.assertEquals(Assert.java:120)
        at org.apache.iceberg.flink.TestChangeLogTable.testSqlChangeLog(TestChangeLogTable.java:285)
        at org.apache.iceberg.flink.TestChangeLogTable.testChangeLogOnDataKey(TestChangeLogTable.java:167)

[hililiwei]

apache#3834

[rdblue]

Description needed

[rdblue]

It's best to avoid personal pronouns in comments and docs. It isn't clear who "we" is and omitting it is typically much more direct. For example: "The columns that are not equality ID columns may have changed in the inserted row, so the delete file must be written using only the equality ID columns."