[flink] partial-update merge engine support batch deletion using flink sql

paimon-common/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1942,6 +1942,10 @@ public List<String> sequenceField() {
  .orElse(Collections.emptyList());
  }
 
+ public boolean partialUpdateRemoveRecordOnDelete() {
+ return options.get(PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE);
+ }
+
  public Optional<String> rowkindField() {
  return options.getOptional(ROWKIND_FIELD);
  }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/SupportsRowLevelOperationFlinkTableSink.java

@@ -185,11 +185,17 @@ private void validateDeletable() {
  table.getClass().getName()));
  }
 
- MergeEngine mergeEngine = CoreOptions.fromMap(table.options()).mergeEngine();
- if (mergeEngine != DEDUPLICATE) {
- throw new UnsupportedOperationException(
- String.format("Merge engine %s can not support batch delete.", mergeEngine));
+ CoreOptions coreOptions = CoreOptions.fromMap(table.options());
+ if (coreOptions.mergeEngine() == DEDUPLICATE
+ || (coreOptions.mergeEngine() == PARTIAL_UPDATE
+ && coreOptions.partialUpdateRemoveRecordOnDelete())) {
+ return;
  }
+
+ throw new UnsupportedOperationException(
+ String.format(
+ "Merge engine %s can not support batch delete.",
+ coreOptions.mergeEngine()));
  }
 
  private boolean canPushDownDeleteFilter() {

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

@@ -621,4 +621,29 @@ public void testIgnoreDelete(boolean localMerge) throws Exception {
  Row.ofKind(RowKind.UPDATE_AFTER, 1, "A", "apple"));
  iterator.close();
  }
+
+ @Test
+ public void testRemoveRecordOnDelete() {
+ 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'"
+ + ")");
+
+ sql("INSERT INTO remove_record_on_delete VALUES (1, CAST (NULL AS STRING), 'apple')");
+
+ // delete record
+ sql("DELETE FROM remove_record_on_delete WHERE pk = 1");
+
+ // batch read
+ assertThat(sql("SELECT * FROM remove_record_on_delete")).isEmpty();
+
+ // insert records
+ sql("INSERT INTO remove_record_on_delete VALUES (1, CAST (NULL AS STRING), 'apache')");
+ sql("INSERT INTO remove_record_on_delete VALUES (1, 'A', CAST (NULL AS STRING))");
+
+ // batch read
+ assertThat(sql("SELECT * FROM remove_record_on_delete"))
+ .containsExactlyInAnyOrder(Row.of(1, "A", "apache"));
+ }
 }