add flink support FLIP-510

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java

@@ -493,6 +493,17 @@ public class FlinkConnectorOptions {
                     .withDescription(
                             "Enable sink writer coordinator to plan data files in Job Manager.");
 
+    public static final ConfigOption<Boolean> SINK_KEY_ONLY_DELETES_ENABLED =
+            key("sink.key-only-deletes.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "If true, a primary-key table sink advertises the FLIP-510 key-only "
+                                    + "(partial) deletes capability, allowing the Flink planner to drop the "
+                                    + "upstream ChangelogNormalize node when the source produces deletes by key. "
+                                    + "This changes the execution plan and requires Flink 2.1+; it has no "
+                                    + "effect on Flink 1.x. Disabled by default.");
+
     public static final ConfigOption<MemorySize> SINK_WRITER_COORDINATOR_CACHE_MEMORY =
             key("sink.writer-coordinator.cache-memory")
                     .memoryType()

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

@@ -21,7 +21,9 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.CoreOptions.ChangelogProducer;
 import org.apache.paimon.CoreOptions.MergeEngine;
+import org.apache.paimon.flink.FlinkConnectorOptions;
 import org.apache.paimon.flink.PaimonDataStreamSinkProvider;
+import org.apache.paimon.flink.utils.ChangelogModeUtils;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.table.FormatTable;
 import org.apache.paimon.table.Table;
@@ -95,6 +97,14 @@ && new CoreOptions(options).definedAggFunc()) {
                     builder.addContainedKind(kind);
                 }
             }
+            // FLIP-510: a primary-key Paimon table applies changes (including deletes) by primary
+            // key, so it can accept key-only (partial) deletes. Advertising this as a sink
+            // capability lets the planner drop the upstream ChangelogNormalize when the source
+            // produces deletes by key. This changes the execution plan, so it is opt-in via
+            // 'sink.key-only-deletes.enabled' and is a no-op on Flink 1.x (API added in Flink 2.1).
+            if (options.get(FlinkConnectorOptions.SINK_KEY_ONLY_DELETES_ENABLED)) {
+                ChangelogModeUtils.enableKeyOnlyDeletes(builder);
+            }
             return builder.build();
         }
     }

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

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.flink.sink.FlinkTableSink;
 import org.apache.paimon.flink.source.DataTableSource;
+import org.apache.paimon.flink.utils.ChangelogModeUtils;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.options.Options;
@@ -89,6 +90,21 @@ public void testDefault() throws Exception {
                         .build());
     }
 
+    @Test
+    public void testKeyOnlyDeletesEnabled() throws Exception {
+        Options options = new Options();
+        options.set(FlinkConnectorOptions.SINK_KEY_ONLY_DELETES_ENABLED, true);
+        // keyOnlyDeletes is a no-op on Flink 1.x and set on Flink 2.1+, so build the expected
+        // mode through the same adapter the sink uses.
+        ChangelogMode.Builder expectSink =
+                ChangelogMode.newBuilder()
+                        .addContainedKind(RowKind.INSERT)
+                        .addContainedKind(RowKind.UPDATE_AFTER)
+                        .addContainedKind(RowKind.DELETE);
+        ChangelogModeUtils.enableKeyOnlyDeletes(expectSink);
+        test(options, ChangelogMode.upsert(), expectSink.build());
+    }
+
     @Test
     public void testInputChangelogProducer() throws Exception {
         Options options = new Options();

paimon-flink/paimon-flink1-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.utils;
+
+import org.apache.flink.table.connector.ChangelogMode;
+
+/** Utility methods about Flink {@link ChangelogMode} to resolve compatibility issues. */
+public class ChangelogModeUtils {
+
+    /** FLIP-510 key-only deletes: not available before Flink 2.1, so this is a no-op. */
+    public static ChangelogMode.Builder enableKeyOnlyDeletes(ChangelogMode.Builder builder) {
+        return builder;
+    }
+}

paimon-flink/paimon-flink2-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.utils;
+
+import org.apache.flink.table.connector.ChangelogMode;
+
+/** Utility methods about Flink {@link ChangelogMode} to resolve compatibility issues. */
+public class ChangelogModeUtils {
+
+    /** FLIP-510 key-only deletes, available from Flink 2.1. */
+    public static ChangelogMode.Builder enableKeyOnlyDeletes(ChangelogMode.Builder builder) {
+        return builder.keyOnlyDeletes(true);
+    }
+}