(paimon) branch master updated: [flink] fix delete record exception after partial-update merge engine configuration to lookup changelog producer (#4345)

[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 a5b1236c2 [flink] fix delete record exception after partial-update 
merge engine configuration to lookup changelog producer (#4345)
a5b1236c2 is described below

commit a5b1236c281054d518c60ed0b3e440509a6bb5bd
Author: Kerwin <37063904+zhuangch...@users.noreply.github.com>
AuthorDate: Tue Oct 29 13:56:54 2024 +0800

    [flink] fix delete record exception after partial-update merge engine 
configuration to lookup changelog producer (#4345)
---
 .../mergetree/compact/PartialUpdateMergeFunction.java  |  8 +++-----
 .../org/apache/paimon/flink/PartialUpdateITCase.java   | 18 ++++++++++++++++--
 2 files changed, 19 insertions(+), 7 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 17d2d937a..11661851c 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
@@ -120,7 +120,6 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
                     currentDeleteRow = true;
                     row = new GenericRow(getters.length);
                 }
-                // ignore -U records
                 return;
             }
 
@@ -254,10 +253,9 @@ public class PartialUpdateMergeFunction implements 
MergeFunction<KeyValue> {
         if (reused == null) {
             reused = new KeyValue();
         }
-        if (currentDeleteRow) {
-            return null;
-        }
-        return reused.replace(currentKey, latestSequenceNumber, 
RowKind.INSERT, row);
+
+        RowKind rowKind = currentDeleteRow ? RowKind.DELETE : RowKind.INSERT;
+        return reused.replace(currentKey, latestSequenceNumber, rowKind, row);
     }
 
     public static MergeFunctionFactory<KeyValue> factory(
diff --git 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
index 68f109f04..e0372b74d 100644
--- 
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
+++ 
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
@@ -623,11 +623,12 @@ public class PartialUpdateITCase extends 
CatalogITCaseBase {
     }
 
     @Test
-    public void testRemoveRecordOnDelete() {
+    public void testRemoveRecordOnDelete() throws Exception {
         sql(
                 "CREATE TABLE remove_record_on_delete (pk INT PRIMARY KEY NOT 
ENFORCED, a STRING, b STRING) WITH ("
                         + " 'merge-engine' = 'partial-update',"
-                        + " 'partial-update.remove-record-on-delete' = 'true'"
+                        + " 'partial-update.remove-record-on-delete' = 'true',"
+                        + " 'changelog-producer' = 'lookup'"
                         + ")");
 
         sql("INSERT INTO remove_record_on_delete VALUES (1, CAST (NULL AS 
STRING), 'apple')");
@@ -645,5 +646,18 @@ public class PartialUpdateITCase extends CatalogITCaseBase 
{
         // batch read
         assertThat(sql("SELECT * FROM remove_record_on_delete"))
                 .containsExactlyInAnyOrder(Row.of(1, "A", "apache"));
+
+        // streaming read results has -U
+        BlockingIterator<Row, Row> iterator =
+                streamSqlBlockIter(
+                        "SELECT * FROM remove_record_on_delete /*+ 
OPTIONS('scan.timestamp-millis' = '0') */");
+        assertThat(iterator.collect(5))
+                .containsExactly(
+                        Row.ofKind(RowKind.INSERT, 1, null, "apple"),
+                        Row.ofKind(RowKind.DELETE, 1, null, "apple"),
+                        Row.ofKind(RowKind.INSERT, 1, null, "apache"),
+                        Row.ofKind(RowKind.UPDATE_BEFORE, 1, null, "apache"),
+                        Row.ofKind(RowKind.UPDATE_AFTER, 1, "A", "apache"));
+        iterator.close();
     }
 }