(paimon) branch master updated: [flink] fix same primary key cannot be inserted again after deletion in Partial Update merge engine. (#4248)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 21bfe4274 [flink] fix same primary key cannot be inserted again after 
deletion in Partial Update merge engine. (#4248)
21bfe4274 is described below

commit 21bfe42742e05f28949191adc7de2533b651105e
Author: Kerwin <37063904+zhuangch...@users.noreply.github.com>
AuthorDate: Thu Sep 26 14:34:41 2024 +0800

    [flink] fix same primary key cannot be inserted again after deletion in 
Partial Update merge engine. (#4248)
---
 .../paimon/mergetree/compact/PartialUpdateMergeFunction.java   | 10 +++++++---
 .../org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java  |  9 ++++++++-
 2 files changed, 15 insertions(+), 4 deletions(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index 09549ec35..17d2d937a 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -73,6 +73,7 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
     private long latestSequenceNumber;
     private GenericRow row;
     private KeyValue reused;
+    private boolean currentDeleteRow;
 
     protected PartialUpdateMergeFunction(
             InternalRow.FieldGetter[] getters,
@@ -100,6 +101,7 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
     public void add(KeyValue kv) {
         // refresh key object to avoid reference overwritten
         currentKey = kv.key();
+        currentDeleteRow = false;
 
         if (kv.valueKind().isRetract()) {
             // In 0.7- versions, the delete records might be written into data 
file even when
@@ -115,7 +117,8 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
 
             if (removeRecordOnDelete) {
                 if (kv.valueKind() == RowKind.DELETE) {
-                    row = null;
+                    currentDeleteRow = true;
+                    row = new GenericRow(getters.length);
                 }
                 // ignore -U records
                 return;
@@ -127,7 +130,8 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
                             "By default, Partial update can not accept delete 
records,"
                                     + " you can choose one of the following 
solutions:",
                             "1. Configure 'ignore-delete' to ignore delete 
records.",
-                            "2. Configure 'sequence-group's to retract partial 
columns.");
+                            "2. Configure 
'partial-update.remove-record-on-delete' to remove the whole row when receiving 
delete records.",
+                            "3. Configure 'sequence-group's to retract partial 
columns.");
 
             throw new IllegalArgumentException(msg);
         }
@@ -250,7 +254,7 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
         if (reused == null) {
             reused = new KeyValue();
         }
-        if (removeRecordOnDelete && row == null) {
+        if (currentDeleteRow) {
             return null;
         }
         return reused.replace(currentKey, latestSequenceNumber, 
RowKind.INSERT, row);
diff --git 
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
 
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
index dc11424a1..7edc41818 100644
--- 
a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
+++ 
b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
@@ -1004,7 +1004,7 @@ public class PrimaryKeyFileStoreTableTest extends 
FileStoreTableTestBase {
         TableRead read = table.newRead();
         StreamTableWrite write = table.newWrite("");
         StreamTableCommit commit = table.newCommit("");
-        // 1. inserts
+        // 1. Inserts
         write.write(GenericRow.of(1, 1, 3, 3));
         write.write(GenericRow.of(1, 1, 1, 1));
         write.write(GenericRow.of(1, 1, 2, 2));
@@ -1030,6 +1030,13 @@ public class PrimaryKeyFileStoreTableTest extends 
FileStoreTableTestBase {
         commit.commit(3, write.prepareCommit(true, 3));
         result = getResult(read, toSplits(snapshotReader.read().dataSplits()), 
rowToString);
         assertThat(result).isEmpty();
+
+        // 5. Inserts
+        write.write(GenericRow.of(1, 1, 2, 2));
+        commit.commit(4, write.prepareCommit(true, 4));
+        result = getResult(read, toSplits(snapshotReader.read().dataSplits()), 
rowToString);
+        assertThat(result).containsExactlyInAnyOrder("+I[1, 1, 2, 2]");
+
         write.close();
         commit.close();
     }