This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 558765aa0a [flink] Support FLIP-510 key-only deletes on primary-key
table sink (#8452)
558765aa0a is described below
commit 558765aa0aaec9552142ae601cf35e0da5e14eac
Author: IRC <[email protected]>
AuthorDate: Wed Jul 15 02:39:28 2026 +0200
[flink] Support FLIP-510 key-only deletes on primary-key table sink (#8452)
---
docs/generated/flink_connector_configuration.html | 6 +
.../apache/paimon/flink/FlinkConnectorOptions.java | 14 ++
.../paimon/flink/sink/FlinkTableSinkBase.java | 25 +++
.../org/apache/paimon/flink/ChangelogModeTest.java | 25 +++
.../apache/paimon/flink/KeyOnlyDeletesITCase.java | 183 +++++++++++++++++++++
.../paimon/flink/utils/ChangelogModeUtils.java | 30 ++++
.../paimon/flink/utils/ChangelogModeUtils.java | 56 +++++++
7 files changed, 339 insertions(+)
diff --git a/docs/generated/flink_connector_configuration.html
b/docs/generated/flink_connector_configuration.html
index a80c819be0..83e9c0bb25 100644
--- a/docs/generated/flink_connector_configuration.html
+++ b/docs/generated/flink_connector_configuration.html
@@ -278,6 +278,12 @@ under the License.
<td>MemorySize</td>
<td>Weight of managed memory for RocksDB in cross-partition
update, Flink will compute the memory size according to the weight, the actual
memory used depends on the running environment.</td>
</tr>
+ <tr>
+ <td><h5>sink.key-only-deletes.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>If true, a primary-key table sink advertises the key-only
(partial) deletes capability, allowing the Flink planner to drop the upstream
ChangelogNormalize node when the source produces deletes by key. Requires Flink
2.1+; no effect on Flink 1.x or 2.0. Does not apply when the table has no
primary key, when 'changelog-producer' is 'input', or when 'merge-engine' is
'aggregation' or 'partial-update' with aggregation functions; in those cases a
warning is logged. Disabled by [...]
+ </tr>
<tr>
<td><h5>sink.managed.writer-buffer-memory</h5></td>
<td style="word-wrap: break-word;">256 mb</td>
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
index 7799acc2d2..bc61b8bbab 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java
@@ -504,6 +504,20 @@ 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
key-only (partial) "
+ + "deletes capability, allowing the Flink
planner to drop the "
+ + "upstream ChangelogNormalize node when
the source produces "
+ + "deletes by key. Requires Flink 2.1+; no
effect on Flink 1.x or 2.0. "
+ + "Does not apply when the table has no
primary key, when "
+ + "'changelog-producer' is 'input', or
when 'merge-engine' is "
+ + "'aggregation' or 'partial-update' with
aggregation functions; "
+ + "in those cases a warning is logged.
Disabled by default.");
+
public static final ConfigOption<MemorySize>
SINK_WRITER_COORDINATOR_CACHE_MEMORY =
key("sink.writer-coordinator.cache-memory")
.memoryType()
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
index 2790a8a926..da5b8b58eb 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
@@ -21,7 +21,9 @@ package org.apache.paimon.flink.sink;
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;
@@ -34,6 +36,8 @@ import
org.apache.flink.table.connector.sink.abilities.SupportsOverwrite;
import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.types.RowKind;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
@@ -57,6 +61,8 @@ public abstract class FlinkTableSinkBase
protected final Table table;
+ private static final Logger LOG =
LoggerFactory.getLogger(FlinkTableSinkBase.class);
+
protected Map<String, String> staticPartitions = new HashMap<>();
protected boolean overwrite = false;
@@ -72,19 +78,24 @@ public abstract class FlinkTableSinkBase
if (table.primaryKeys().isEmpty()) {
// Don't check this, for example, only inserts are available from
the database, but the
// plan phase contains all changelogs
+ warnKeyOnlyDeletesIgnored("the table has no primary key");
return requestedMode;
} else {
Options options = Options.fromMap(table.options());
if (options.get(CHANGELOG_PRODUCER) == ChangelogProducer.INPUT) {
+ warnKeyOnlyDeletesIgnored("'changelog-producer' is 'input'");
return requestedMode;
}
if (options.get(MERGE_ENGINE) == MergeEngine.AGGREGATE) {
+ warnKeyOnlyDeletesIgnored("'merge-engine' is 'aggregation'");
return requestedMode;
}
if (options.get(MERGE_ENGINE) == MergeEngine.PARTIAL_UPDATE
&& new CoreOptions(options).definedAggFunc()) {
+ warnKeyOnlyDeletesIgnored(
+ "'merge-engine' is 'partial-update' with aggregation
functions");
return requestedMode;
}
@@ -95,10 +106,24 @@ public abstract class FlinkTableSinkBase
builder.addContainedKind(kind);
}
}
+ if
(options.get(FlinkConnectorOptions.SINK_KEY_ONLY_DELETES_ENABLED)) {
+ ChangelogModeUtils.enableKeyOnlyDeletes(builder);
+ }
return builder.build();
}
}
+ private void warnKeyOnlyDeletesIgnored(String reason) {
+ if (Options.fromMap(table.options())
+ .get(FlinkConnectorOptions.SINK_KEY_ONLY_DELETES_ENABLED)) {
+ LOG.warn(
+ "'{}' is set to true for table {}, but it has no effect
because {}.",
+ FlinkConnectorOptions.SINK_KEY_ONLY_DELETES_ENABLED.key(),
+ tableIdentifier.asSummaryString(),
+ reason);
+ }
+ }
+
@Override
public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
if (overwrite && !context.isBounded()) {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
index b968df4746..6a068b2ee3 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ChangelogModeTest.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink;
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 class ChangelogModeTest {
.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();
@@ -96,6 +112,15 @@ public class ChangelogModeTest {
test(options, ChangelogMode.all(), ChangelogMode.all());
}
+ @Test
+ public void testKeyOnlyDeletesIgnoredWithInputChangelogProducer() throws
Exception {
+ Options options = new Options();
+ options.set(CHANGELOG_PRODUCER, INPUT);
+ options.set(FlinkConnectorOptions.SINK_KEY_ONLY_DELETES_ENABLED, true);
+ // The input changelog producer forces ChangelogMode.all(), so
key-only deletes is ignored.
+ test(options, ChangelogMode.all(), ChangelogMode.all());
+ }
+
@Test
public void testLookupChangelogProducer() throws Exception {
Options options = new Options();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/KeyOnlyDeletesITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/KeyOnlyDeletesITCase.java
new file mode 100644
index 0000000000..dc8a5052dd
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/KeyOnlyDeletesITCase.java
@@ -0,0 +1,183 @@
+/*
+ * 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;
+
+import org.apache.paimon.utils.BlockingIterator;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
+
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.types.Row;
+import org.apache.flink.types.RowKind;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * ITCase for FLIP-510 key-only (partial) deletes on a primary-key table with
{@code merge-engine =
+ * deduplicate}, {@code changelog-producer = none} and {@code
sink.key-only-deletes.enabled = true}.
+ *
+ * <p>Verifies data integrity end-to-end when the upstream source emits:
+ *
+ * <ul>
+ * <li>updates as {@code +U} only (no {@code -U}/UPDATE_BEFORE), and
+ * <li>deletes carrying only the primary key ({@code -D} with the value
columns null).
+ * </ul>
+ *
+ * <p>The delete-by-key and update-without-before shapes are guaranteed by how
the input is built (a
+ * {@code values} source with {@code changelog-mode = 'I,UA,D'} and DELETE
rows populated only on
+ * the key). Enabling {@code sink.key-only-deletes.enabled} lets the Flink
planner drop the upstream
+ * ChangelogNormalize so those records reach the Paimon sink as-is. The API
backing this capability
+ * only exists in Flink 2.1+, so the test is gated on that.
+ */
+public class KeyOnlyDeletesITCase extends CatalogITCaseBase {
+
+ /**
+ * Streams +I / +U (no -U) / key-only -D over two keys:
+ *
+ * <ul>
+ * <li>k1: +I (1,1) → +U (1,2) → -D (1) — deleted, must be gone;
+ * <li>k2: +I (2,20) → -D (2) → +I (2,21) — deleted then re-inserted,
latest value kept.
+ * </ul>
+ *
+ * <p>Final state must be exactly (2,21), proving the key-only delete
removes the row and leaves
+ * no tombstone stuck to the key on re-insert.
+ */
+ @Test
+ @EnabledIf("isFlink2_1OrAbove")
+ public void testKeyOnlyDeletesIntegrity() throws Exception {
+ sql(
+ "CREATE TABLE T (k INT PRIMARY KEY NOT ENFORCED, v INT) WITH ("
+ + " 'merge-engine' = 'deduplicate',"
+ + " 'changelog-producer' = 'none',"
+ + " 'sink.key-only-deletes.enabled' = 'true')");
+
+ List<Row> input =
+ ImmutableList.of(
+ Row.ofKind(RowKind.INSERT, 1, 1), // +I (k1, v1)
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, 2), // +U (k1,
v2), no -U
+ Row.ofKind(RowKind.DELETE, 1, null), // -D (k1),
key-only → gone
+ Row.ofKind(RowKind.INSERT, 2, 20), // +I (k2, v20)
+ Row.ofKind(RowKind.DELETE, 2, null), // -D (k2),
key-only
+ Row.ofKind(RowKind.INSERT, 2, 21)); // +I (k2, v21),
re-insert
+
+ String id = TestValuesTableFactory.registerData(input);
+ streamSqlIter(
+ "CREATE TEMPORARY TABLE input (k INT PRIMARY KEY NOT
ENFORCED, v INT) WITH ("
+ + " 'connector' = 'values',"
+ + " 'bounded' = 'true',"
+ + " 'data-id' = '%s',"
+ + " 'changelog-mode' = 'I,UA,D')",
+ id)
+ .close();
+ sEnv.executeSql("INSERT INTO T SELECT * FROM input").await();
+
+ assertThat(sql("SELECT * FROM T")).containsExactlyInAnyOrder(Row.of(2,
21));
+ }
+
+ private static Stream<Arguments> reconstructParameters() {
+ // parameters: mergeEngine, changelogProducer
+ return Stream.of(
+ Arguments.of("deduplicate", "lookup"),
+ Arguments.of("deduplicate", "full-compaction"),
+ Arguments.of("partial-update", "lookup"),
+ Arguments.of("partial-update", "full-compaction"));
+ }
+
+ /**
+ * Same key-only-delete / no-before-update write path as {@link
#testKeyOnlyDeletesIntegrity()},
+ * but with a changelog-producer ({@code lookup} / {@code
full-compaction}) that materializes a
+ * changelog. Uses a partial schema (k, a, b) so partial-update exercises
real column merging.
+ *
+ * <p>Input over key 1: {@code +I (1,1,1)} → {@code +U (1,2,1)} (no {@code
-U}) → {@code -D (1)}
+ * (key-only, value columns null). Each phase is written in its own commit
(multiple changes to
+ * the same key within a single commit would be merged before the producer
runs, so the
+ * intermediate transition would not surface). Two verifications:
+ *
+ * <ul>
+ * <li>Batch read: final state is empty (the key-only delete removed the
row).
+ * <li>Streaming read: even though the write was key-only / before-less,
the producer
+ * reconstructs a full changelog — the update surfaces as a {@code
-U}/{@code +U} pair
+ * carrying the before-image, and the delete surfaces as a {@code
-D} with the full row
+ * (all columns), not just the key.
+ * </ul>
+ */
+ @ParameterizedTest(name = "mergeEngine={0}, changelogProducer={1}")
+ @MethodSource("reconstructParameters")
+ @EnabledIf("isFlink2_1OrAbove")
+ public void testKeyOnlyDeletesReconstructChangelog(String mergeEngine,
String changelogProducer)
+ throws Exception {
+ sql(
+ "CREATE TABLE R (k INT PRIMARY KEY NOT ENFORCED, a INT, b INT)
WITH ("
+ + " 'merge-engine' = '%s',"
+ + " 'partial-update.remove-record-on-delete' = 'true',"
+ + " 'changelog-producer' = '%s',"
+ + " 'changelog-producer.compaction-interval' = '1s',"
+ + " 'sink.key-only-deletes.enabled' = 'true')",
+ mergeEngine, changelogProducer);
+
+ // Open the streaming changelog read before writing so we observe the
reconstructed stream.
+ BlockingIterator<Row, Row> iterator =
BlockingIterator.of(streamSqlIter("SELECT * FROM R"));
+
+ // Commit 1: insert.
+ streamInto(ImmutableList.of(Row.ofKind(RowKind.INSERT, 1, 1, 1))); //
+I (1, a=1, b=1)
+
assertThat(iterator.collect(1)).containsExactly(Row.ofKind(RowKind.INSERT, 1,
1, 1));
+
+ // Commit 2: update as +U only (no -U); reconstructed as a -U/+U pair
with the before-image.
+ streamInto(ImmutableList.of(Row.ofKind(RowKind.UPDATE_AFTER, 1, 2,
1))); // +U (1, a=2)
+ assertThat(iterator.collect(2))
+ .containsExactlyInAnyOrder(
+ Row.ofKind(RowKind.UPDATE_BEFORE, 1, 1, 1),
+ Row.ofKind(RowKind.UPDATE_AFTER, 1, 2, 1));
+
+ // Commit 3: key-only delete (only the primary key is populated).
+ streamInto(ImmutableList.of(Row.ofKind(RowKind.DELETE, 1, null,
null)));
+
+ // The reconstructed delete carries the full row (all columns), not
just the key.
+
assertThat(iterator.collect(1)).containsExactly(Row.ofKind(RowKind.DELETE, 1,
2, 1));
+ iterator.close();
+
+ // Batch read: the key-only delete removed the row.
+ assertThat(sql("SELECT * FROM R")).isEmpty();
+ }
+
+ /**
+ * Streams {@code rows} into table {@code R} through a bounded {@code
values} changelog source
+ * that carries only INSERT / UPDATE_AFTER / DELETE (no UPDATE_BEFORE).
+ */
+ private void streamInto(List<Row> rows) throws Exception {
+ String id = TestValuesTableFactory.registerData(rows);
+ streamSqlIter(
+ "CREATE TEMPORARY TABLE src_%s (k INT PRIMARY KEY NOT
ENFORCED, a INT, b INT) WITH ("
+ + " 'connector' = 'values',"
+ + " 'bounded' = 'true',"
+ + " 'data-id' = '%s',"
+ + " 'changelog-mode' = 'I,UA,D')",
+ id, id)
+ .close();
+ sEnv.executeSql(String.format("INSERT INTO R SELECT * FROM src_%s",
id)).await();
+ }
+}
diff --git
a/paimon-flink/paimon-flink1-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java
b/paimon-flink/paimon-flink1-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java
new file mode 100644
index 0000000000..4aece06d44
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/paimon-flink/paimon-flink2-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java
b/paimon-flink/paimon-flink2-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java
new file mode 100644
index 0000000000..7798c5b551
--- /dev/null
+++
b/paimon-flink/paimon-flink2-common/src/main/java/org/apache/paimon/flink/utils/ChangelogModeUtils.java
@@ -0,0 +1,56 @@
+/*
+ * 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.runtime.util.EnvironmentInformation;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Utility methods about Flink {@link ChangelogMode} to resolve compatibility
issues. */
+public class ChangelogModeUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(ChangelogModeUtils.class);
+
+ public static ChangelogMode.Builder
enableKeyOnlyDeletes(ChangelogMode.Builder builder) {
+ if (isFlink21OrAbove()) {
+ return builder.keyOnlyDeletes(true);
+ }
+ LOG.warn(
+ "'sink.key-only-deletes.enabled' requires Flink 2.1+ (current
version: {}); "
+ + "key-only deletes are not enabled.",
+ EnvironmentInformation.getVersion());
+ return builder;
+ }
+
+ private static boolean isFlink21OrAbove() {
+ String version = EnvironmentInformation.getVersion();
+ try {
+ String[] parts = version.split("\\.");
+ int major = Integer.parseInt(parts[0]);
+ int minor = Integer.parseInt(parts[1]);
+ return major > 2 || (major == 2 && minor >= 1);
+ } catch (RuntimeException e) {
+ LOG.warn(
+ "Could not parse Flink version '{}'; key-only deletes are
not enabled.",
+ version);
+ return false;
+ }
+ }
+}
