(paimon) branch master updated: [flink] Fix InsertOnly behavior for CDC and deduplication scenarios (#4294)

[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 b51b7cfd2 [flink] Fix InsertOnly behavior for CDC and deduplication 
scenarios (#4294)
b51b7cfd2 is described below

commit b51b7cfd2de022b2e1de752a57974bad67d994ef
Author: yunfengzhou-hub <yuri.zhouyunf...@outlook.com>
AuthorDate: Thu Oct 10 15:34:34 2024 +0800

    [flink] Fix InsertOnly behavior for CDC and deduplication scenarios (#4294)
---
 .../java/org/apache/paimon/mergetree/MergeTreeWriter.java   |  3 +++
 .../org/apache/paimon/flink/sink/StoreSinkWriteImpl.java    | 13 +++++++++++--
 2 files changed, 14 insertions(+), 2 deletions(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java 
b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index de0a28c33..b73be6941 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -289,6 +289,9 @@ public class MergeTreeWriter implements 
RecordWriter<KeyValue>, MemoryOwner {
 
     @Override
     public void withInsertOnly(boolean insertOnly) {
+        if (this.isInsertOnly == insertOnly) {
+            return;
+        }
         if (insertOnly && writeBuffer != null && writeBuffer.size() > 0) {
             throw new IllegalStateException(
                     "Insert-only can only be set before any record is 
received.");
diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
index bdaf7bc32..709c2be63 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreSinkWriteImpl.java
@@ -65,6 +65,8 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
 
     @Nullable private final MetricGroup metricGroup;
 
+    @Nullable private Boolean lastInsertOnly;
+
     public StoreSinkWriteImpl(
             FileStoreTable table,
             String commitUser,
@@ -154,15 +156,21 @@ public class StoreSinkWriteImpl implements StoreSinkWrite 
{
         }
 
         if (memoryPoolFactory != null) {
-            return tableWrite.withMemoryPoolFactory(memoryPoolFactory);
+            tableWrite.withMemoryPoolFactory(memoryPoolFactory);
         } else {
-            return tableWrite.withMemoryPool(
+            tableWrite.withMemoryPool(
                     memoryPool != null
                             ? memoryPool
                             : new HeapMemorySegmentPool(
                                     table.coreOptions().writeBufferSize(),
                                     table.coreOptions().pageSize()));
         }
+
+        if (lastInsertOnly != null) {
+            tableWrite.withInsertOnly(lastInsertOnly);
+        }
+
+        return tableWrite;
     }
 
     public void withCompactExecutor(ExecutorService compactExecutor) {
@@ -171,6 +179,7 @@ public class StoreSinkWriteImpl implements StoreSinkWrite {
 
     @Override
     public void withInsertOnly(boolean insertOnly) {
+        this.lastInsertOnly = insertOnly;
         write.withInsertOnly(insertOnly);
     }