diff --git a/docs/docs/flink-configuration.md b/docs/docs/flink-configuration.md index 2f70cbf576d8..90d5fcc3b8a0 100644 --- a/docs/docs/flink-configuration.md +++ b/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`. diff --git a/docs/docs/flink-maintenance.md b/docs/docs/flink-maintenance.md index 6f5975a5e4ed..638dfb172c84 100644 --- a/docs/docs/flink-maintenance.md +++ b/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 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'; diff --git a/docs/docs/flink-writes.md b/docs/docs/flink-writes.md index 393d5ac563b1..6ffedd5a77ae 100644 --- a/docs/docs/flink-writes.md +++ b/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: diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java index fd3fccb224a2..ca5c566bfa1b 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java +++ b/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 diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java index 1fdd6df8d753..6ad4bf30bb45 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java +++ b/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 CONVERT_EQUALITY_DELETES_ENABLE = + ConfigOptions.key(ConvertEqualityDeletesConfig.PREFIX + "enabled") + .booleanType() + .defaultValue(false); + @Experimental public static final ConfigOption TABLE_REFRESH_INTERVAL = ConfigOptions.key("table-refresh-interval").durationType().noDefaultValue(); diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java index f51df3338d02..4b4316fa085b 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletes.java +++ b/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 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 append(DataStream trigger) { Preconditions.checkNotNull(stagingBranch, "stagingBranch must be set"); diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/ConvertEqualityDeletesConfig.java new file mode 100644 index 000000000000..71b7c546fd38 --- /dev/null +++ b/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 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 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 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 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.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java index 34d7330c5913..206c03725940 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/FlinkMaintenanceConfig.java +++ b/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); } diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java index 440fdb278be8..70f8c63e63bd 100644 --- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java +++ b/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 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 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 equalityFieldColumnsSet = equalityFieldColumns != null ? Sets.newHashSet(equalityFieldColumns) : null; @@ -814,6 +845,32 @@ IcebergSink build() { equalityFieldColumnsSet); } + private void addConvertEqualityDeletesTask( + FlinkWriteConf flinkWriteConf, + FlinkMaintenanceConfig flinkMaintenanceConfig, + Set 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 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.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletesConfig.java new file mode 100644 index 000000000000..f83f67f79e18 --- /dev/null +++ b/flink/v2.1/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 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.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java index 5c926d7c25d5..049589f3a4d0 100644 --- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java +++ b/flink/v2.1/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.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.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 flinkConf; @BeforeEach @@ -164,6 +178,15 @@ private void setupDeleteOrphanFilesConfig() { 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 @@ public void testAllMaintenanceE2e(String lockType) throws Exception { 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 rows = Lists.newArrayList(Row.of(1, "hello")); + DataStream 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 rows = Lists.newArrayList(Row.of(1, "hello")); + DataStream 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 rows = Lists.newArrayList(Row.of(1, "bbb")); + DataStream 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 reader = + ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { + for (DeleteFile file : reader) { + if (ContentFileUtil.isDV(file)) { + count++; + } + } + } + } + + return count; + } }