Flink: Integrate ConvertEqualityDeletes with IcebergSink

docs/docs/flink-configuration.md

@@ -163,6 +163,20 @@ INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
 | shred-variants                          | Table write.parquet.shred-variants         | Overrides this table's shred variants for this write |
 | variant-inference-buffer-size           | Table write.parquet.variant-inference-buffer-size | Overrides this table's variant inference buffer size for this write |
 
+#### Post-commit maintenance options
+
+`IcebergSink` can run [table maintenance](flink-maintenance.md#icebergsink-with-post-commit-integration)
+after each commit. The enable flags are listed below; see the maintenance docs for the full per-task
+option set.
+
+| Flink option | Default | Description |
+|--------------|---------|-------------|
+| flink-maintenance.rewrite.enabled | false | Compact data files after commit |
+| flink-maintenance.expire-snapshots.enabled | false | Expire snapshots after commit |
+| flink-maintenance.delete-orphan-files.enabled | false | Delete orphan files after commit |
+| flink-maintenance.convert-equality-deletes.enabled | false | Convert equality deletes to deletion vectors after commit (requires equality fields and format version >= 3) |
+| flink-maintenance.convert-equality-deletes.target-branch | Write branch | Branch the converted DVs are committed to; defaults to the write branch (in-place conversion) |
+
 #### Range distribution statistics type
 
 Config value is a enum type: `Map`, `Sketch`, `Auto`.

docs/docs/flink-maintenance.md

@@ -375,6 +375,18 @@ IcebergSink.forRowData(dataStream)
     .append();
 ```
 
+`convertEqualityDeletes()` converts equality deletes to deletion vectors. Unlike the other tasks, it requires equality field columns (upsert or CDC writes) and a table with format version >= 3. By default, it converts in place on the write branch:
+
+```java
+IcebergSink.forRowData(dataStream)
+    .table(table)
+    .tableLoader(tableLoader)
+    .upsert(true)
+    .equalityFieldColumns(List.of("id"))
+    .convertEqualityDeletes()
+    .append();
+```
+
 ##### Config
 
 All maintenance tasks are configured through string properties:
@@ -386,6 +398,8 @@ Map<String, String> flinkConf = new HashMap<>();
 flinkConf.put("flink-maintenance.rewrite.enabled", "true");
 flinkConf.put("flink-maintenance.expire-snapshots.enabled", "true");
 flinkConf.put("flink-maintenance.delete-orphan-files.enabled", "true");
+// Requires equality field columns (upsert or CDC) and a table with format version >= 3.
+flinkConf.put("flink-maintenance.convert-equality-deletes.enabled", "true");
 
 // Configure rewrite data files
 flinkConf.put("flink-maintenance.rewrite.max-bytes", "1073741824");
@@ -397,6 +411,10 @@ flinkConf.put("flink-maintenance.expire-snapshots.max-snapshot-age-seconds", "60
 // Configure delete orphan files
 flinkConf.put("flink-maintenance.delete-orphan-files.min-age-seconds", "259200");
 
+// Configure convert equality deletes. Converts in place on the write branch by default;
+// set a target branch to instead promote the converted deletion vectors there.
+flinkConf.put("flink-maintenance.convert-equality-deletes.target-branch", "main");
+
 // Configure JDBC lock settings (deprecated, lock configuration is no longer required for a single Flink job)
 flinkConf.put("flink-maintenance.lock.type", "jdbc");
 flinkConf.put("flink-maintenance.lock.jdbc.uri", "jdbc:postgresql://localhost:5432/iceberg");
@@ -419,6 +437,8 @@ SET 'table.exec.iceberg.use.v2.sink' = 'true';
 SET 'flink-maintenance.rewrite.enabled' = 'true';
 SET 'flink-maintenance.expire-snapshots.enabled' = 'true';
 SET 'flink-maintenance.delete-orphan-files.enabled' = 'true';
+-- Requires equality field columns (upsert) and a table with format version >= 3
+SET 'flink-maintenance.convert-equality-deletes.enabled' = 'true';
 
 -- Configure rewrite data files
 SET 'flink-maintenance.rewrite.max-bytes' = '1073741824';

docs/docs/flink-writes.md

@@ -398,6 +398,32 @@ To use SinkV2 based implementation, replace `FlinkSink` with `IcebergSink` in th
      - The `RANGE` distribution mode is not yet available for the `IcebergSink`
      - When using `IcebergSink` use `uidSuffix` instead of the `uidPrefix`
 
+### Post-commit table maintenance
+
+`IcebergSink` can run [table maintenance](flink-maintenance.md#icebergsink-with-post-commit-integration)
+tasks automatically after each commit, so a separate maintenance job is not required. Enable them with
+builder methods (or the equivalent `flink-maintenance.*` write options):
+
+- `rewriteDataFiles()` compacts small data files
+- `expireSnapshots()` expires old snapshots
+- `deleteOrphanFiles()` removes unreferenced files
+- `convertEqualityDeletes()` converts equality deletes to deletion vectors (requires equality fields
+  and table format version >= 3)
+
+```java
+IcebergSink.forRowData(dataStream)
+    .table(table)
+    .tableLoader(tableLoader)
+    .upsert(true)
+    .equalityFieldColumns(List.of("id"))
+    .rewriteDataFiles()
+    .convertEqualityDeletes()
+    .append();
+```
+
+See [IcebergSink with Post-Commit Integration](flink-maintenance.md#icebergsink-with-post-commit-integration)
+for the full options, locking, and the `convertEqualityDeletes` staging/target-branch setup.
+
 ## Flink Dynamic Iceberg Sink
 
 The Flink Dynamic Iceberg Sink (Dynamic Sink) allows:

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java

@@ -247,6 +247,15 @@ public boolean compactMode() {
         .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

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java

@@ -23,6 +23,7 @@
 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 @@ private 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();

flink/v2.1/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 Builder equalityFieldColumns(List<String> columns) {
       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");

flink/v2.1/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();
+  }
+}

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java

@@ -128,6 +128,10 @@ public DeleteOrphanFilesConfig createDeleteOrphanFilesConfig() {
     return new DeleteOrphanFilesConfig(table, writeProperties, readableConfig);
   }
 
+  public ConvertEqualityDeletesConfig createConvertEqualityDeletesConfig() {
+    return new ConvertEqualityDeletesConfig(table, writeProperties, readableConfig);
+  }
+
   public LockConfig createLockConfig() {
     return new LockConfig(table, writeProperties, readableConfig);
   }

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java

@@ -73,6 +73,8 @@
 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 Builder deleteOrphanFiles(Map<String, String> config) {
       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 @@ IcebergSink build() {
         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 @@ IcebergSink build() {
           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.
      *