(incubator-paimon) branch master updated: [flink] Fix DELETE exception with lookup changelog-producer and pk as delete key (#2584)

[lz...@apache.org]

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/incubator-paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new a6c5a6763 [flink] Fix DELETE exception with lookup changelog-producer and pk as delete key (#2584)
a6c5a6763 is described below

commit a6c5a6763139da23d8f761da3848b1d51aef6730
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Thu Dec 28 19:04:30 2023 +0800

    [flink] Fix DELETE exception with lookup changelog-producer and pk as delete key (#2584)
---
 .../main/java/org/apache/paimon/table/TableUtils.java   |  7 +++++++
 .../org/apache/paimon/flink/BatchFileStoreITCase.java   | 17 +++++++++++++++++
 2 files changed, 24 insertions(+)

diff --git a/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
index 00bf5541d..91460c1bd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
@@ -19,6 +19,8 @@
 package org.apache.paimon.table;
 
 import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.disk.IOManagerImpl;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.predicate.PredicateFilter;
 import org.apache.paimon.reader.RecordReader;
@@ -35,6 +37,8 @@ import java.util.List;
 /** Utils for Table. TODO we can introduce LocalAction maybe? */
 public class TableUtils {
 
+    private static final String TEMP_DIR = System.getProperty("java.io.tmpdir");
+
     /**
      * Delete according to filters.
      *
@@ -49,7 +53,10 @@ public class TableUtils {
         long hit = 0;
         try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(splits);
                 BatchTableWrite write = writeBuilder.newWrite();
+                // we create temp io manager for writer
+                IOManager ioManager = new IOManagerImpl(TEMP_DIR);
                 BatchTableCommit commit = writeBuilder.newCommit()) {
+            write.withIOManager(ioManager);
             CloseableIterator<InternalRow> iterator = reader.toCloseableIterator();
             PredicateFilter filter = new PredicateFilter(table.rowType(), filters);
             while (iterator.hasNext()) {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index 0a1e06473..211fe1411 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -388,6 +388,23 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
         iterator.close();
     }
 
+    @Test
+    public void testDeleteWithPkLookup() throws Exception {
+        sql(
+                "CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED, v STRING) "
+                        + "WITH ('changelog-producer' = 'lookup')");
+        BlockingIterator<Row, Row> iterator = streamSqlBlockIter("SELECT * FROM ignore_delete");
+
+        sql("INSERT INTO ignore_delete VALUES (1, 'A'), (2, 'B')");
+        sql("DELETE FROM ignore_delete WHERE pk = 1");
+        sql("INSERT INTO ignore_delete VALUES (1, 'B')");
+
+        assertThat(iterator.collect(2))
+                .containsExactlyInAnyOrder(
+                        Row.ofKind(RowKind.INSERT, 1, "B"), Row.ofKind(RowKind.INSERT, 2, "B"));
+        iterator.close();
+    }
+
     @Test
     public void testScanFromOldSchema() throws InterruptedException {
         sql("CREATE TABLE select_old (f0 INT PRIMARY KEY NOT ENFORCED, f1 STRING)");