[core] Validate retract records in PostponeBucketWriter

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/KeyValueFileStore.java

@@ -173,6 +173,7 @@ public AbstractFileStoreWrite<KeyValue> newWrite(String commitUser, @Nullable In
                     partitionType,
                     keyType,
                     valueType,
+                    mfFactory,
                     this::pathFactory,
                     newReaderFactoryBuilder(),
                     snapshotManager(),

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeFunction.java

@@ -48,5 +48,6 @@ public interface MergeFunction<T> {
     /** Get current merged value. */
     T getResult();
 
+    /** Require copy input kv, this may cache kv in memory. */
     boolean requireCopy();
 }

paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketFileStoreWrite.java

@@ -29,6 +29,7 @@
 import org.apache.paimon.io.KeyValueFileReaderFactory;
 import org.apache.paimon.io.KeyValueFileWriterFactory;
 import org.apache.paimon.mergetree.compact.ConcatRecordReader;
+import org.apache.paimon.mergetree.compact.MergeFunctionFactory;
 import org.apache.paimon.operation.FileStoreScan;
 import org.apache.paimon.operation.FileStoreWrite;
 import org.apache.paimon.operation.MemoryFileStoreWrite;
@@ -71,6 +72,7 @@ public class PostponeBucketFileStoreWrite extends MemoryFileStoreWrite<KeyValue>
     private final KeyValueFileWriterFactory.Builder writerFactoryBuilder;
     private final FileIO fileIO;
     private final FileStorePathFactory pathFactory;
+    private final MergeFunctionFactory<KeyValue> mfFactory;
     private final KeyValueFileReaderFactory.Builder readerFactoryBuilder;
 
     private boolean forceBufferSpill = false;
@@ -83,6 +85,7 @@ public PostponeBucketFileStoreWrite(
             RowType partitionType,
             RowType keyType,
             RowType valueType,
+            MergeFunctionFactory<KeyValue> mfFactory,
             BiFunction<CoreOptions, String, FileStorePathFactory> formatPathFactory,
             KeyValueFileReaderFactory.Builder readerFactoryBuilder,
             SnapshotManager snapshotManager,
@@ -93,6 +96,7 @@ public PostponeBucketFileStoreWrite(
         super(snapshotManager, scan, options, partitionType, null, null, tableName);
         this.fileIO = fileIO;
         this.pathFactory = pathFactory;
+        this.mfFactory = mfFactory;
         this.readerFactoryBuilder = readerFactoryBuilder;
 
         Options newOptions = new Options(options.toMap());
@@ -189,6 +193,7 @@ protected PostponeBucketWriter createWriter(
                 options.spillCompressOptions(),
                 options.writeBufferSpillDiskSize(),
                 ioManager,
+                mfFactory.create(),
                 writerFactory,
                 files -> newFileRead(partition, bucket, files),
                 forceBufferSpill,

paimon-core/src/main/java/org/apache/paimon/postpone/PostponeBucketWriter.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.manifest.FileSource;
 import org.apache.paimon.memory.MemoryOwner;
 import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.mergetree.compact.MergeFunction;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.reader.RecordReaderIterator;
 import org.apache.paimon.types.RowType;
@@ -55,6 +56,7 @@ public class PostponeBucketWriter implements RecordWriter<KeyValue>, MemoryOwner
 
     private final FileIO fileIO;
     private final DataFilePathFactory pathFactory;
+    private final MergeFunction<KeyValue> mergeFunction;
     private final KeyValueFileWriterFactory writerFactory;
     private final List<DataFileMeta> files;
     private final IOFunction<List<DataFileMeta>, RecordReaderIterator<KeyValue>> fileRead;
@@ -71,12 +73,14 @@ public PostponeBucketWriter(
             CompressOptions spillCompression,
             MemorySize maxDiskSize,
             @Nullable IOManager ioManager,
+            MergeFunction<KeyValue> mergeFunction,
             KeyValueFileWriterFactory writerFactory,
             IOFunction<List<DataFileMeta>, RecordReaderIterator<KeyValue>> fileRead,
             boolean useWriteBuffer,
             boolean spillable,
             @Nullable CommitIncrement restoreIncrement) {
         this.ioManager = ioManager;
+        this.mergeFunction = mergeFunction;
         this.writerFactory = writerFactory;
         this.fileRead = fileRead;
         this.fileIO = fileIO;
@@ -99,6 +103,7 @@ private RollingFileWriter<KeyValue, DataFileMeta> createRollingRowWriter() {
 
     @Override
     public void write(KeyValue record) throws Exception {
+        validateRetract(record);
         boolean success = sinkWriter.write(record);
         if (!success) {
             flush();
@@ -112,6 +117,14 @@ public void write(KeyValue record) throws Exception {
         }
     }
 
+    private void validateRetract(KeyValue kv) {
+        if (kv.valueKind().isRetract()) {
+            mergeFunction.reset();
+            mergeFunction.add(kv);
+            mergeFunction.getResult();
+        }
+    }
+
     private void flush() throws Exception {
         files.addAll(sinkWriter.flush());
     }

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PostponeBucketTableITCase.java

@@ -35,12 +35,47 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** IT cases for postpone bucket tables. */
 public class PostponeBucketTableITCase extends AbstractTestBase {
 
     private static final int TIMEOUT = 120;
 
+    @Test
+    public void testRetractOnPartialUpdate() {
+        String warehouse = getTempDirPath();
+        TableEnvironment tEnv =
+                tableEnvironmentBuilder()
+                        .batchMode()
+                        .setConf(TableConfigOptions.TABLE_DML_SYNC, true)
+                        .build();
+
+        tEnv.executeSql(
+                "CREATE CATALOG mycat WITH (\n"
+                        + "  'type' = 'paimon',\n"
+                        + "  'warehouse' = '"
+                        + warehouse
+                        + "'\n"
+                        + ")");
+        tEnv.executeSql("USE CATALOG mycat");
+        tEnv.executeSql(
+                "CREATE TABLE T (\n"
+                        + "  k INT,\n"
+                        + "  v1 INT,\n"
+                        + "  v2 INT,\n"
+                        + "  row_kind_col STRING,\n"
+                        + "  PRIMARY KEY (k) NOT ENFORCED\n"
+                        + ") WITH (\n"
+                        + "  'bucket' = '-2',\n"
+                        + "  'merge-engine' = 'partial-update',\n"
+                        + "  'rowkind.field' = 'row_kind_col'\n"
+                        + ")");
+        assertThatThrownBy(() -> tEnv.executeSql("INSERT INTO T VALUES (1, 1, 1, '-D')").await())
+                .rootCause()
+                .hasMessageContaining("By default, Partial update can not accept delete records");
+    }
+
     @Test
     public void testWriteThenCompact() throws Exception {
         String warehouse = getTempDirPath();