You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by hu...@apache.org on 2026/07/10 22:46:35 UTC
(iceberg) branch main updated: Flink: Backport: Integrate ConvertEqualityDeletes with IcebergSink (#17142) (#17156)
This is an automated email from the ASF dual-hosted git repository.
huaxingao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 9ddc391c1a Flink: Backport: Integrate ConvertEqualityDeletes with IcebergSink (#17142) (#17156)
9ddc391c1a is described below
commit 9ddc391c1ac863e94be6520be5adc63ccdc19eb9
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Sat Jul 11 00:46:28 2026 +0200
Flink: Backport: Integrate ConvertEqualityDeletes with IcebergSink (#17142) (#17156)
---
.../org/apache/iceberg/flink/FlinkWriteConf.java | 9 ++
.../apache/iceberg/flink/FlinkWriteOptions.java | 6 +
.../maintenance/api/ConvertEqualityDeletes.java | 17 +++
.../api/ConvertEqualityDeletesConfig.java | 90 ++++++++++++++
.../maintenance/api/FlinkMaintenanceConfig.java | 4 +
.../org/apache/iceberg/flink/sink/IcebergSink.java | 57 +++++++++
.../api/TestConvertEqualityDeletesConfig.java | 69 +++++++++++
.../sink/TestIcebergSinkTableMaintenance.java | 138 +++++++++++++++++++++
.../org/apache/iceberg/flink/FlinkWriteConf.java | 9 ++
.../apache/iceberg/flink/FlinkWriteOptions.java | 6 +
.../maintenance/api/ConvertEqualityDeletes.java | 17 +++
.../api/ConvertEqualityDeletesConfig.java | 90 ++++++++++++++
.../maintenance/api/FlinkMaintenanceConfig.java | 4 +
.../org/apache/iceberg/flink/sink/IcebergSink.java | 57 +++++++++
.../api/TestConvertEqualityDeletesConfig.java | 69 +++++++++++
.../sink/TestIcebergSinkTableMaintenance.java | 138 +++++++++++++++++++++
16 files changed, 780 insertions(+)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java
index 990d23f2aa..055672d29e 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java
@@ -247,6 +247,15 @@ public class FlinkWriteConf {
.parse();
}
+ public boolean convertEqualityDeletesMode() {
+ return confParser
+ .booleanConf()
+ .option(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.key())
+ .flinkConfig(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE)
+ .defaultValue(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.defaultValue())
+ .parse();
+ }
+
/**
* NOTE: This may be removed or changed in a future release. This value specifies the interval for
* refreshing the table instances in sink writer subtasks. If not specified then the default
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java
index ee2aeaa450..8a868d41f5 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.Experimental;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletesConfig;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFilesConfig;
import org.apache.iceberg.flink.maintenance.api.ExpireSnapshotsConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
@@ -98,6 +99,11 @@ public class FlinkWriteOptions {
.booleanType()
.defaultValue(false);
+ public static final ConfigOption<Boolean> CONVERT_EQUALITY_DELETES_ENABLE =
+ ConfigOptions.key(ConvertEqualityDeletesConfig.PREFIX + "enabled")
+ .booleanType()
+ .defaultValue(false);
+
@Experimental
public static final ConfigOption<Duration> TABLE_REFRESH_INTERVAL =
ConfigOptions.key("table-refresh-interval").durationType().noDefaultValue();
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java
index f51df3338d..4b4316fa08 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.flink.maintenance.api;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -126,6 +127,22 @@ public class ConvertEqualityDeletes {
return this;
}
+ /**
+ * Configures the scheduling of the conversion.
+ *
+ * @param config properties for the conversion, see {@link ConvertEqualityDeletesConfig} for
+ * available keys
+ */
+ public Builder config(ConvertEqualityDeletesConfig config) {
+ scheduleOnCommitCount(config.scheduleOnCommitCount());
+ Long intervalSecond = config.scheduleOnIntervalSecond();
+ if (intervalSecond != null) {
+ scheduleOnInterval(Duration.ofSeconds(intervalSecond));
+ }
+
+ return this;
+ }
+
@Override
DataStream<TaskResult> append(DataStream<Trigger> trigger) {
Preconditions.checkNotNull(stagingBranch, "stagingBranch must be set");
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java
new file mode 100644
index 0000000000..71b7c546fd
--- /dev/null
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java
@@ -0,0 +1,90 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.FlinkConfParser;
+
+public class ConvertEqualityDeletesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "convert-equality-deletes.";
+
+ public static final String TARGET_BRANCH = PREFIX + "target-branch";
+ public static final ConfigOption<String> TARGET_BRANCH_OPTION =
+ ConfigOptions.key(TARGET_BRANCH)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The branch where converted data files and deletion vectors are committed. "
+ + "Defaults to the sink's write branch, i.e. an in-place conversion.");
+
+ public static final String SCHEDULE_ON_COMMIT_COUNT = PREFIX + "schedule.commit-count";
+ public static final ConfigOption<Integer> SCHEDULE_ON_COMMIT_COUNT_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_COMMIT_COUNT)
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "The number of commits after which to trigger a new equality delete conversion.");
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "The time interval (in seconds) between two consecutive equality delete conversions.");
+
+ private final FlinkConfParser confParser;
+
+ public ConvertEqualityDeletesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ /** Gets the target branch, or null to convert in place on the sink's write branch. */
+ public String targetBranch() {
+ return confParser
+ .stringConf()
+ .option(TARGET_BRANCH)
+ .flinkConfig(TARGET_BRANCH_OPTION)
+ .parseOptional();
+ }
+
+ /** Gets the number of commits that trigger a conversion. */
+ public int scheduleOnCommitCount() {
+ return confParser
+ .intConf()
+ .option(SCHEDULE_ON_COMMIT_COUNT)
+ .flinkConfig(SCHEDULE_ON_COMMIT_COUNT_OPTION)
+ .defaultValue(SCHEDULE_ON_COMMIT_COUNT_OPTION.defaultValue())
+ .parse();
+ }
+
+ /** Gets the time interval (in seconds) between conversions, or null when not set. */
+ public Long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .parseOptional();
+ }
+}
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java
index 34d7330c59..206c037259 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java
@@ -128,6 +128,10 @@ public class FlinkMaintenanceConfig {
return new DeleteOrphanFilesConfig(table, writeProperties, readableConfig);
}
+ public ConvertEqualityDeletesConfig createConvertEqualityDeletesConfig() {
+ return new ConvertEqualityDeletesConfig(table, writeProperties, readableConfig);
+ }
+
public LockConfig createLockConfig() {
return new LockConfig(table, writeProperties, readableConfig);
}
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
index d5f3114d0b..18e44da496 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
@@ -72,6 +72,8 @@ import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.flink.FlinkWriteConf;
import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletes;
+import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletesConfig;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFiles;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFilesConfig;
import org.apache.iceberg.flink.maintenance.api.ExpireSnapshots;
@@ -715,6 +717,31 @@ public class IcebergSink
return this;
}
+ /**
+ * Enables converting equality deletes to deletion vectors as a post-commit maintenance task.
+ * The sink's write branch is read for equality delete files, which are converted to deletion
+ * vectors and committed back to the same branch (or the target branch, if configured). Requires
+ * equality field columns (upsert or CDC) and a table with format version >= 3.
+ *
+ * @see ConvertEqualityDeletesConfig for the default config.
+ */
+ public Builder convertEqualityDeletes() {
+ writeOptions.put(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables converting equality deletes to deletion vectors as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ConvertEqualityDeletesConfig} for
+ * available keys.
+ */
+ public Builder convertEqualityDeletes(Map<String, String> config) {
+ convertEqualityDeletes();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
writeOptions.put(FlinkWriteOptions.BRANCH.key(), branch);
@@ -792,6 +819,10 @@ public class IcebergSink
maintenanceTasks.add(DeleteOrphanFiles.builder().config(deleteOrphanFilesConfig));
}
+ if (flinkWriteConf.convertEqualityDeletesMode()) {
+ addConvertEqualityDeletesTask(flinkWriteConf, flinkMaintenanceConfig, equalityFieldIds);
+ }
+
Set<String> equalityFieldColumnsSet =
equalityFieldColumns != null ? Sets.newHashSet(equalityFieldColumns) : null;
@@ -814,6 +845,32 @@ public class IcebergSink
equalityFieldColumnsSet);
}
+ private void addConvertEqualityDeletesTask(
+ FlinkWriteConf flinkWriteConf,
+ FlinkMaintenanceConfig flinkMaintenanceConfig,
+ Set<Integer> equalityFieldIds) {
+ Preconditions.checkState(
+ !equalityFieldIds.isEmpty(),
+ "Equality field columns must be set to convert equality deletes to deletion vectors.");
+ ConvertEqualityDeletesConfig convertEqualityDeletesConfig =
+ flinkMaintenanceConfig.createConvertEqualityDeletesConfig();
+ // The sink writes equality deletes to its write branch, so that becomes the staging branch.
+ // The target defaults to the same branch, i.e. an in-place conversion.
+ String targetBranch = convertEqualityDeletesConfig.targetBranch();
+ String convertTargetBranch = targetBranch != null ? targetBranch : flinkWriteConf.branch();
+ List<String> convertEqualityFieldColumns = Lists.newArrayList();
+ for (int fieldId : equalityFieldIds) {
+ convertEqualityFieldColumns.add(table.schema().findColumnName(fieldId));
+ }
+
+ maintenanceTasks.add(
+ ConvertEqualityDeletes.builder()
+ .stagingBranch(flinkWriteConf.branch())
+ .targetBranch(convertTargetBranch)
+ .equalityFieldColumns(convertEqualityFieldColumns)
+ .config(convertEqualityDeletesConfig));
+ }
+
/**
* Append the iceberg sink operators to write records to iceberg table.
*
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java
new file mode 100644
index 0000000000..f83f67f79e
--- /dev/null
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.flink.configuration.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestConvertEqualityDeletesConfig extends OperatorTestBase {
+ private Table table;
+ private final Map<String, String> input = Maps.newHashMap();
+
+ @BeforeEach
+ public void before() {
+ this.table = createTable();
+ input.put(ConvertEqualityDeletesConfig.TARGET_BRANCH, "myTarget");
+ input.put(ConvertEqualityDeletesConfig.SCHEDULE_ON_COMMIT_COUNT, "5");
+ input.put(ConvertEqualityDeletesConfig.SCHEDULE_ON_INTERVAL_SECOND, "120");
+ input.put("other.config", "should-be-ignored");
+ }
+
+ @AfterEach
+ public void after() {
+ input.clear();
+ }
+
+ @Test
+ void testConfigParsing() {
+ ConvertEqualityDeletesConfig config =
+ new ConvertEqualityDeletesConfig(table, input, new Configuration());
+
+ assertThat(config.targetBranch()).isEqualTo("myTarget");
+ assertThat(config.scheduleOnCommitCount()).isEqualTo(5);
+ assertThat(config.scheduleOnIntervalSecond()).isEqualTo(120L);
+ }
+
+ @Test
+ void testConfigDefaults() {
+ ConvertEqualityDeletesConfig config =
+ new ConvertEqualityDeletesConfig(table, Maps.newHashMap(), new Configuration());
+
+ assertThat(config.targetBranch()).isNull();
+ assertThat(config.scheduleOnCommitCount()).isEqualTo(1);
+ assertThat(config.scheduleOnIntervalSecond()).isNull();
+ }
+}
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
index 5c926d7c25..049589f3a4 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
@@ -19,11 +19,15 @@
package org.apache.iceberg.flink.sink;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -31,13 +35,17 @@ import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.Row;
import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
+import org.apache.iceberg.data.GenericAppenderHelper;
import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.MiniFlinkClusterExtension;
import org.apache.iceberg.flink.SimpleDataUtil;
@@ -47,15 +55,21 @@ import org.apache.iceberg.flink.maintenance.api.ExpireSnapshotsConfig;
import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.FieldSource;
class TestIcebergSinkTableMaintenance extends TestFlinkIcebergSinkBase {
private static final String[] LOCK_TYPES = new String[] {LockConfig.JdbcLockConfig.JDBC, ""};
+ @TempDir private Path tempDir;
+
private Map<String, String> flinkConf;
@BeforeEach
@@ -164,6 +178,15 @@ class TestIcebergSinkTableMaintenance extends TestFlinkIcebergSinkBase {
flinkConf.put(DeleteOrphanFilesConfig.MIN_AGE_SECONDS, "86400");
}
+ private void setupConvertEqualityDeletesConfig() {
+ flinkConf.put(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.key(), "true");
+ }
+
+ private void upgradeToFormatV3() {
+ table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+ table.refresh();
+ }
+
private void setupLockConfig(String lockType) {
if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
@@ -318,4 +341,119 @@ class TestIcebergSinkTableMaintenance extends TestFlinkIcebergSinkBase {
getDataFiles(table.snapshot(table.currentSnapshot().parentId()), table);
assertThat(preCompactDataFiles).hasSize(3);
}
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesOperatorAdded(String lockType) {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "hello"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ boolean containsConvert = false;
+ for (JobVertex vertex : streamGraph.getJobGraph().getVertices()) {
+ if (vertex.getName().contains("EqConvert")) {
+ containsConvert = true;
+ break;
+ }
+ }
+
+ assertThat(containsConvert).isTrue();
+ }
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesRequiresEqualityFields(String lockType) {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "hello"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ // No equality field columns and no identifier fields on the schema, so there is nothing to
+ // convert.
+ assertThatThrownBy(
+ () ->
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .setAll(flinkConf)
+ .append())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Equality field columns must be set");
+ }
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesE2e(String lockType) throws Exception {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+
+ // Pre-existing row on main. The sink then upserts the same key, writing an equality delete that
+ // removes this row, which the converter resolves to a deletion vector in a single cycle.
+ new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
+ .appendToTable(ImmutableList.of(SimpleDataUtil.createRecord(1, "aaa")));
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "bbb"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ JobClient jobClient = env.executeAsync("Test Convert Equality Deletes E2E");
+ try {
+ Awaitility.await()
+ .atMost(Duration.ofMinutes(2))
+ .pollInterval(Duration.ofMillis(200))
+ .untilAsserted(() -> assertThat(dvCountOnMain(table)).isEqualTo(1));
+ SimpleDataUtil.assertTableRecords(
+ table, ImmutableList.of(SimpleDataUtil.createRecord(1, "bbb")));
+ } finally {
+ jobClient.cancel();
+ }
+ }
+
+ private static long dvCountOnMain(Table table) throws IOException {
+ table.refresh();
+ if (table.currentSnapshot() == null) {
+ return 0;
+ }
+
+ long count = 0;
+ for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile file : reader) {
+ if (ContentFileUtil.isDV(file)) {
+ count++;
+ }
+ }
+ }
+ }
+
+ return count;
+ }
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java
index 990d23f2aa..055672d29e 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java
@@ -247,6 +247,15 @@ public class FlinkWriteConf {
.parse();
}
+ public boolean convertEqualityDeletesMode() {
+ return confParser
+ .booleanConf()
+ .option(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.key())
+ .flinkConfig(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE)
+ .defaultValue(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.defaultValue())
+ .parse();
+ }
+
/**
* NOTE: This may be removed or changed in a future release. This value specifies the interval for
* refreshing the table instances in sink writer subtasks. If not specified then the default
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java
index ee2aeaa450..8a868d41f5 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.Experimental;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.iceberg.SnapshotRef;
+import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletesConfig;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFilesConfig;
import org.apache.iceberg.flink.maintenance.api.ExpireSnapshotsConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
@@ -98,6 +99,11 @@ public class FlinkWriteOptions {
.booleanType()
.defaultValue(false);
+ public static final ConfigOption<Boolean> CONVERT_EQUALITY_DELETES_ENABLE =
+ ConfigOptions.key(ConvertEqualityDeletesConfig.PREFIX + "enabled")
+ .booleanType()
+ .defaultValue(false);
+
@Experimental
public static final ConfigOption<Duration> TABLE_REFRESH_INTERVAL =
ConfigOptions.key("table-refresh-interval").durationType().noDefaultValue();
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java
index f51df3338d..4b4316fa08 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.flink.maintenance.api;
+import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Set;
@@ -126,6 +127,22 @@ public class ConvertEqualityDeletes {
return this;
}
+ /**
+ * Configures the scheduling of the conversion.
+ *
+ * @param config properties for the conversion, see {@link ConvertEqualityDeletesConfig} for
+ * available keys
+ */
+ public Builder config(ConvertEqualityDeletesConfig config) {
+ scheduleOnCommitCount(config.scheduleOnCommitCount());
+ Long intervalSecond = config.scheduleOnIntervalSecond();
+ if (intervalSecond != null) {
+ scheduleOnInterval(Duration.ofSeconds(intervalSecond));
+ }
+
+ return this;
+ }
+
@Override
DataStream<TaskResult> append(DataStream<Trigger> trigger) {
Preconditions.checkNotNull(stagingBranch, "stagingBranch must be set");
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java
new file mode 100644
index 0000000000..71b7c546fd
--- /dev/null
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java
@@ -0,0 +1,90 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import java.util.Map;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.FlinkConfParser;
+
+public class ConvertEqualityDeletesConfig {
+ public static final String PREFIX = FlinkMaintenanceConfig.PREFIX + "convert-equality-deletes.";
+
+ public static final String TARGET_BRANCH = PREFIX + "target-branch";
+ public static final ConfigOption<String> TARGET_BRANCH_OPTION =
+ ConfigOptions.key(TARGET_BRANCH)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The branch where converted data files and deletion vectors are committed. "
+ + "Defaults to the sink's write branch, i.e. an in-place conversion.");
+
+ public static final String SCHEDULE_ON_COMMIT_COUNT = PREFIX + "schedule.commit-count";
+ public static final ConfigOption<Integer> SCHEDULE_ON_COMMIT_COUNT_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_COMMIT_COUNT)
+ .intType()
+ .defaultValue(1)
+ .withDescription(
+ "The number of commits after which to trigger a new equality delete conversion.");
+
+ public static final String SCHEDULE_ON_INTERVAL_SECOND = PREFIX + "schedule.interval-second";
+ public static final ConfigOption<Long> SCHEDULE_ON_INTERVAL_SECOND_OPTION =
+ ConfigOptions.key(SCHEDULE_ON_INTERVAL_SECOND)
+ .longType()
+ .noDefaultValue()
+ .withDescription(
+ "The time interval (in seconds) between two consecutive equality delete conversions.");
+
+ private final FlinkConfParser confParser;
+
+ public ConvertEqualityDeletesConfig(
+ Table table, Map<String, String> writeOptions, ReadableConfig readableConfig) {
+ this.confParser = new FlinkConfParser(table, writeOptions, readableConfig);
+ }
+
+ /** Gets the target branch, or null to convert in place on the sink's write branch. */
+ public String targetBranch() {
+ return confParser
+ .stringConf()
+ .option(TARGET_BRANCH)
+ .flinkConfig(TARGET_BRANCH_OPTION)
+ .parseOptional();
+ }
+
+ /** Gets the number of commits that trigger a conversion. */
+ public int scheduleOnCommitCount() {
+ return confParser
+ .intConf()
+ .option(SCHEDULE_ON_COMMIT_COUNT)
+ .flinkConfig(SCHEDULE_ON_COMMIT_COUNT_OPTION)
+ .defaultValue(SCHEDULE_ON_COMMIT_COUNT_OPTION.defaultValue())
+ .parse();
+ }
+
+ /** Gets the time interval (in seconds) between conversions, or null when not set. */
+ public Long scheduleOnIntervalSecond() {
+ return confParser
+ .longConf()
+ .option(SCHEDULE_ON_INTERVAL_SECOND)
+ .flinkConfig(SCHEDULE_ON_INTERVAL_SECOND_OPTION)
+ .parseOptional();
+ }
+}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java
index 34d7330c59..206c037259 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java
@@ -128,6 +128,10 @@ public class FlinkMaintenanceConfig {
return new DeleteOrphanFilesConfig(table, writeProperties, readableConfig);
}
+ public ConvertEqualityDeletesConfig createConvertEqualityDeletesConfig() {
+ return new ConvertEqualityDeletesConfig(table, writeProperties, readableConfig);
+ }
+
public LockConfig createLockConfig() {
return new LockConfig(table, writeProperties, readableConfig);
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
index 440fdb278b..70f8c63e63 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
@@ -73,6 +73,8 @@ import org.apache.iceberg.flink.FlinkSchemaUtil;
import org.apache.iceberg.flink.FlinkWriteConf;
import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.TableLoader;
+import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletes;
+import org.apache.iceberg.flink.maintenance.api.ConvertEqualityDeletesConfig;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFiles;
import org.apache.iceberg.flink.maintenance.api.DeleteOrphanFilesConfig;
import org.apache.iceberg.flink.maintenance.api.ExpireSnapshots;
@@ -715,6 +717,31 @@ public class IcebergSink
return this;
}
+ /**
+ * Enables converting equality deletes to deletion vectors as a post-commit maintenance task.
+ * The sink's write branch is read for equality delete files, which are converted to deletion
+ * vectors and committed back to the same branch (or the target branch, if configured). Requires
+ * equality field columns (upsert or CDC) and a table with format version >= 3.
+ *
+ * @see ConvertEqualityDeletesConfig for the default config.
+ */
+ public Builder convertEqualityDeletes() {
+ writeOptions.put(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.key(), "true");
+ return this;
+ }
+
+ /**
+ * Enables converting equality deletes to deletion vectors as a post-commit maintenance task.
+ *
+ * @param config task-specific configuration, see {@link ConvertEqualityDeletesConfig} for
+ * available keys.
+ */
+ public Builder convertEqualityDeletes(Map<String, String> config) {
+ convertEqualityDeletes();
+ writeOptions.putAll(config);
+ return this;
+ }
+
@Override
public Builder toBranch(String branch) {
writeOptions.put(FlinkWriteOptions.BRANCH.key(), branch);
@@ -792,6 +819,10 @@ public class IcebergSink
maintenanceTasks.add(DeleteOrphanFiles.builder().config(deleteOrphanFilesConfig));
}
+ if (flinkWriteConf.convertEqualityDeletesMode()) {
+ addConvertEqualityDeletesTask(flinkWriteConf, flinkMaintenanceConfig, equalityFieldIds);
+ }
+
Set<String> equalityFieldColumnsSet =
equalityFieldColumns != null ? Sets.newHashSet(equalityFieldColumns) : null;
@@ -814,6 +845,32 @@ public class IcebergSink
equalityFieldColumnsSet);
}
+ private void addConvertEqualityDeletesTask(
+ FlinkWriteConf flinkWriteConf,
+ FlinkMaintenanceConfig flinkMaintenanceConfig,
+ Set<Integer> equalityFieldIds) {
+ Preconditions.checkState(
+ !equalityFieldIds.isEmpty(),
+ "Equality field columns must be set to convert equality deletes to deletion vectors.");
+ ConvertEqualityDeletesConfig convertEqualityDeletesConfig =
+ flinkMaintenanceConfig.createConvertEqualityDeletesConfig();
+ // The sink writes equality deletes to its write branch, so that becomes the staging branch.
+ // The target defaults to the same branch, i.e. an in-place conversion.
+ String targetBranch = convertEqualityDeletesConfig.targetBranch();
+ String convertTargetBranch = targetBranch != null ? targetBranch : flinkWriteConf.branch();
+ List<String> convertEqualityFieldColumns = Lists.newArrayList();
+ for (int fieldId : equalityFieldIds) {
+ convertEqualityFieldColumns.add(table.schema().findColumnName(fieldId));
+ }
+
+ maintenanceTasks.add(
+ ConvertEqualityDeletes.builder()
+ .stagingBranch(flinkWriteConf.branch())
+ .targetBranch(convertTargetBranch)
+ .equalityFieldColumns(convertEqualityFieldColumns)
+ .config(convertEqualityDeletesConfig));
+ }
+
/**
* Append the iceberg sink operators to write records to iceberg table.
*
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java
new file mode 100644
index 0000000000..f83f67f79e
--- /dev/null
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java
@@ -0,0 +1,69 @@
+/*
+ * 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.iceberg.flink.maintenance.api;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Map;
+import org.apache.flink.configuration.Configuration;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.flink.maintenance.operator.OperatorTestBase;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class TestConvertEqualityDeletesConfig extends OperatorTestBase {
+ private Table table;
+ private final Map<String, String> input = Maps.newHashMap();
+
+ @BeforeEach
+ public void before() {
+ this.table = createTable();
+ input.put(ConvertEqualityDeletesConfig.TARGET_BRANCH, "myTarget");
+ input.put(ConvertEqualityDeletesConfig.SCHEDULE_ON_COMMIT_COUNT, "5");
+ input.put(ConvertEqualityDeletesConfig.SCHEDULE_ON_INTERVAL_SECOND, "120");
+ input.put("other.config", "should-be-ignored");
+ }
+
+ @AfterEach
+ public void after() {
+ input.clear();
+ }
+
+ @Test
+ void testConfigParsing() {
+ ConvertEqualityDeletesConfig config =
+ new ConvertEqualityDeletesConfig(table, input, new Configuration());
+
+ assertThat(config.targetBranch()).isEqualTo("myTarget");
+ assertThat(config.scheduleOnCommitCount()).isEqualTo(5);
+ assertThat(config.scheduleOnIntervalSecond()).isEqualTo(120L);
+ }
+
+ @Test
+ void testConfigDefaults() {
+ ConvertEqualityDeletesConfig config =
+ new ConvertEqualityDeletesConfig(table, Maps.newHashMap(), new Configuration());
+
+ assertThat(config.targetBranch()).isNull();
+ assertThat(config.scheduleOnCommitCount()).isEqualTo(1);
+ assertThat(config.scheduleOnIntervalSecond()).isNull();
+ }
+}
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
index 5c926d7c25..049589f3a4 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
@@ -19,11 +19,15 @@
package org.apache.iceberg.flink.sink;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.IOException;
+import java.nio.file.Path;
+import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -31,13 +35,17 @@ import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.Row;
import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
+import org.apache.iceberg.data.GenericAppenderHelper;
import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.MiniFlinkClusterExtension;
import org.apache.iceberg.flink.SimpleDataUtil;
@@ -47,15 +55,21 @@ import org.apache.iceberg.flink.maintenance.api.ExpireSnapshotsConfig;
import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
import org.apache.iceberg.flink.util.FlinkCompatibilityUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.ContentFileUtil;
+import org.awaitility.Awaitility;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.FieldSource;
class TestIcebergSinkTableMaintenance extends TestFlinkIcebergSinkBase {
private static final String[] LOCK_TYPES = new String[] {LockConfig.JdbcLockConfig.JDBC, ""};
+ @TempDir private Path tempDir;
+
private Map<String, String> flinkConf;
@BeforeEach
@@ -164,6 +178,15 @@ class TestIcebergSinkTableMaintenance extends TestFlinkIcebergSinkBase {
flinkConf.put(DeleteOrphanFilesConfig.MIN_AGE_SECONDS, "86400");
}
+ private void setupConvertEqualityDeletesConfig() {
+ flinkConf.put(FlinkWriteOptions.CONVERT_EQUALITY_DELETES_ENABLE.key(), "true");
+ }
+
+ private void upgradeToFormatV3() {
+ table.updateProperties().set(TableProperties.FORMAT_VERSION, "3").commit();
+ table.refresh();
+ }
+
private void setupLockConfig(String lockType) {
if (lockType.equals(LockConfig.JdbcLockConfig.JDBC)) {
flinkConf.put(LockConfig.LOCK_TYPE_OPTION.key(), LockConfig.JdbcLockConfig.JDBC);
@@ -318,4 +341,119 @@ class TestIcebergSinkTableMaintenance extends TestFlinkIcebergSinkBase {
getDataFiles(table.snapshot(table.currentSnapshot().parentId()), table);
assertThat(preCompactDataFiles).hasSize(3);
}
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesOperatorAdded(String lockType) {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "hello"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ boolean containsConvert = false;
+ for (JobVertex vertex : streamGraph.getJobGraph().getVertices()) {
+ if (vertex.getName().contains("EqConvert")) {
+ containsConvert = true;
+ break;
+ }
+ }
+
+ assertThat(containsConvert).isTrue();
+ }
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesRequiresEqualityFields(String lockType) {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "hello"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ // No equality field columns and no identifier fields on the schema, so there is nothing to
+ // convert.
+ assertThatThrownBy(
+ () ->
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .setAll(flinkConf)
+ .append())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("Equality field columns must be set");
+ }
+
+ @ParameterizedTest(name = "lockType = {0}")
+ @FieldSource("LOCK_TYPES")
+ public void testConvertEqualityDeletesE2e(String lockType) throws Exception {
+ setupLockConfig(lockType);
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+
+ // Pre-existing row on main. The sink then upserts the same key, writing an equality delete that
+ // removes this row, which the converter resolves to a deletion vector in a single cycle.
+ new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
+ .appendToTable(ImmutableList.of(SimpleDataUtil.createRecord(1, "aaa")));
+
+ List<Row> rows = Lists.newArrayList(Row.of(1, "bbb"));
+ DataStream<RowData> dataStream =
+ env.addSource(createBoundedSource(rows), ROW_TYPE_INFO)
+ .map(CONVERTER::toInternal, FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ JobClient jobClient = env.executeAsync("Test Convert Equality Deletes E2E");
+ try {
+ Awaitility.await()
+ .atMost(Duration.ofMinutes(2))
+ .pollInterval(Duration.ofMillis(200))
+ .untilAsserted(() -> assertThat(dvCountOnMain(table)).isEqualTo(1));
+ SimpleDataUtil.assertTableRecords(
+ table, ImmutableList.of(SimpleDataUtil.createRecord(1, "bbb")));
+ } finally {
+ jobClient.cancel();
+ }
+ }
+
+ private static long dvCountOnMain(Table table) throws IOException {
+ table.refresh();
+ if (table.currentSnapshot() == null) {
+ return 0;
+ }
+
+ long count = 0;
+ for (ManifestFile manifest : table.currentSnapshot().deleteManifests(table.io())) {
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) {
+ for (DeleteFile file : reader) {
+ if (ContentFileUtil.isDV(file)) {
+ count++;
+ }
+ }
+ }
+ }
+
+ return count;
+ }
}