This is an automated email from the ASF dual-hosted git repository.
mxm 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 753c82df65 Flink: Allow configuring a branch for rewrite data files
action (#18661)
753c82df65 is described below
commit 753c82df656bacf516c17ccddc0792a62933aa17
Author: Swapna Marru <[email protected]>
AuthorDate: Thu Nov 5 08:20:28 2026 -0800
Flink: Allow configuring a branch for rewrite data files action (#18661)
---
docs/docs/flink-maintenance.md | 1 +
.../flink/maintenance/api/RewriteDataFiles.java | 4 +
.../maintenance/api/RewriteDataFilesConfig.java | 17 +++++
.../org/apache/iceberg/flink/sink/IcebergSink.java | 39 ++++++++--
.../maintenance/api/TestRewriteDataFiles.java | 35 ++++++++-
.../api/TestRewriteDataFilesConfig.java | 2 +
.../sink/TestIcebergSinkTableMaintenance.java | 86 ++++++++++++++++++++++
7 files changed, 173 insertions(+), 11 deletions(-)
diff --git a/docs/docs/flink-maintenance.md b/docs/docs/flink-maintenance.md
index eac8d84a08..c7b24f2e92 100644
--- a/docs/docs/flink-maintenance.md
+++ b/docs/docs/flink-maintenance.md
@@ -511,6 +511,7 @@ These keys are used in SQL (SET or table WITH options) or
via `IcebergSink.Build
| `flink-maintenance.rewrite.schedule.data-file-count` | Trigger after N data
files | `1000` |
| `flink-maintenance.rewrite.schedule.data-file-size` | Trigger after total
data file size (bytes) | `107374182400` (100GB) |
| `flink-maintenance.rewrite.schedule.interval-second` | Trigger after time
interval (seconds) | `600` |
+| `flink-maintenance.rewrite.branch` | Branch to compact. Defaults to the
sink's write branch, or to the convert equality deletes target branch when
convert equality deletes is also enabled | Not set |
| `flink-maintenance.rewrite.max-bytes` | Maximum bytes to rewrite per
execution | `Long.MAX_VALUE` |
| `flink-maintenance.rewrite.partial-progress.enabled` | Enable partial
progress commits | `false` |
| `flink-maintenance.rewrite.partial-progress.max-commits` | Maximum commits
for partial progress | `10` |
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
index 8d21dc0b6b..2a72cb83f6 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFiles.java
@@ -260,6 +260,10 @@ public class RewriteDataFiles {
.scheduleOnInterval(
Duration.ofSeconds(rewriteDataFilesConfig.scheduleOnIntervalSecond()));
+ if (rewriteDataFilesConfig.branch() != null) {
+ this.branch(rewriteDataFilesConfig.branch());
+ }
+
// override the rewrite options
this.rewriteOptions.putAll(rewriteDataFilesConfig.properties());
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFilesConfig.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFilesConfig.java
index 1a4645cb2b..2d8722f38a 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFilesConfig.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/maintenance/api/RewriteDataFilesConfig.java
@@ -30,6 +30,15 @@ import org.apache.iceberg.util.PropertyUtil;
public class RewriteDataFilesConfig {
public static final String PREFIX = FlinkMaintenanceConfig.PREFIX +
"rewrite.";
+ public static final String BRANCH = PREFIX + "branch";
+ public static final ConfigOption<String> BRANCH_OPTION =
+ ConfigOptions.key(BRANCH)
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "The branch to compact. Defaults to the sink's write branch, or
to the convert "
+ + "equality deletes target branch when convert equality
deletes is also enabled.");
+
public static final String MAX_BYTES = PREFIX + "max-bytes";
public static final ConfigOption<Long> MAX_BYTES_OPTION =
ConfigOptions.key(MAX_BYTES)
@@ -170,6 +179,14 @@ public class RewriteDataFilesConfig {
.parse();
}
+ /**
+ * Gets the branch to compact, or null to default to the sink's write
branch, or to the convert
+ * equality deletes target branch when convert equality deletes is also
enabled.
+ */
+ public String branch() {
+ return
confParser.stringConf().option(BRANCH).flinkConfig(BRANCH_OPTION).parseOptional();
+ }
+
public Map<String, String> properties() {
return PropertyUtil.propertiesWithPrefix(writeProperties, PREFIX);
}
diff --git
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
index 789bb00082..b436c969db 100644
---
a/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
+++
b/flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/sink/IcebergSink.java
@@ -803,10 +803,17 @@ public class IcebergSink
if (flinkWriteConf.compactMode()) {
RewriteDataFilesConfig rewriteDataFilesConfig =
flinkMaintenanceConfig.createRewriteDataFilesConfig();
+
+ String defaultRewriteBranch = flinkWriteConf.branch();
+ if (flinkWriteConf.convertEqualityDeletesMode()) {
+ defaultRewriteBranch =
+ convertTargetBranch(
+ flinkWriteConf,
flinkMaintenanceConfig.createConvertEqualityDeletesConfig());
+ validateRewriteBranch(defaultRewriteBranch, rewriteDataFilesConfig);
+ }
+
maintenanceTasks.add(
- RewriteDataFiles.builder()
- .branch(flinkWriteConf.branch())
- .config(rewriteDataFilesConfig));
+
RewriteDataFiles.builder().branch(defaultRewriteBranch).config(rewriteDataFilesConfig));
}
if (flinkWriteConf.expireSnapshotsMode()) {
@@ -856,10 +863,8 @@ public class IcebergSink
"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();
+ String convertTargetBranch =
+ convertTargetBranch(flinkWriteConf, convertEqualityDeletesConfig);
List<String> convertEqualityFieldColumns = Lists.newArrayList();
for (int fieldId : equalityFieldIds) {
convertEqualityFieldColumns.add(table.schema().findColumnName(fieldId));
@@ -873,6 +878,26 @@ public class IcebergSink
.config(convertEqualityDeletesConfig));
}
+ private String convertTargetBranch(
+ FlinkWriteConf flinkWriteConf, ConvertEqualityDeletesConfig
convertEqualityDeletesConfig) {
+ // 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();
+ return targetBranch != null ? targetBranch : flinkWriteConf.branch();
+ }
+
+ private void validateRewriteBranch(
+ String convertTargetBranch, RewriteDataFilesConfig
rewriteDataFilesConfig) {
+ String rewriteBranch = rewriteDataFilesConfig.branch();
+ Preconditions.checkArgument(
+ rewriteBranch == null || rewriteBranch.equals(convertTargetBranch),
+ "Invalid rewrite branch %s. Expected rewrite branch %s, "
+ + "the convert equality deletes target branch. Unset config %s
to default to it.",
+ rewriteBranch,
+ convertTargetBranch,
+ RewriteDataFilesConfig.BRANCH);
+ }
+
/**
* Append the iceberg sink operators to write records to iceberg table.
*
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
index 7cb389b663..3ea0c3e2f6 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFiles.java
@@ -33,6 +33,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.time.Instant;
import java.util.List;
import java.util.stream.StreamSupport;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.ManifestFiles;
@@ -50,6 +51,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.FieldSource;
class TestRewriteDataFiles extends MaintenanceTaskTestBase {
+ private static final String TEST_BRANCH = "test-branch";
private static final FileFormat[] FILE_FORMATS =
new FileFormat[] {FileFormat.AVRO, FileFormat.PARQUET, FileFormat.ORC};
@@ -619,18 +621,43 @@ class TestRewriteDataFiles extends
MaintenanceTaskTestBase {
@Test
void testBranch() throws Exception {
+ Table table = createTableWithBranch(TEST_BRANCH);
+
+ rewriteAndAssertBranchCompacted(
+ table,
RewriteDataFiles.builder().rewriteAll(true).branch(TEST_BRANCH));
+ }
+
+ @Test
+ void testRewriteOnCustomBranch() throws Exception {
+ Table table = createTableWithBranch(TEST_BRANCH);
+ RewriteDataFilesConfig config =
+ new RewriteDataFilesConfig(
+ table,
+ ImmutableMap.of(RewriteDataFilesConfig.BRANCH, TEST_BRANCH),
+ new Configuration());
+
+ rewriteAndAssertBranchCompacted(
+ table,
+
RewriteDataFiles.builder().rewriteAll(true).branch(SnapshotRef.MAIN_BRANCH).config(config));
+ }
+
+ /** Creates a table with a branch at two data files, while main gets a third
one. */
+ private Table createTableWithBranch(String branchName) throws Exception {
Table table = createTable();
insert(table, 1, "a");
insert(table, 2, "b");
// Create branch based on above inserts
- String branchName = "test-branch";
table.manageSnapshots().createBranch(branchName).commit();
// Insert another file on main only (main has 3 files, branch stays at 2)
insert(table, 3, "c");
+ return table;
+ }
-
appendRewriteDataFiles(RewriteDataFiles.builder().rewriteAll(true).branch(branchName));
+ private void rewriteAndAssertBranchCompacted(Table table,
RewriteDataFiles.Builder builder)
+ throws Exception {
+ appendRewriteDataFiles(builder);
runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
@@ -638,7 +665,7 @@ class TestRewriteDataFiles extends MaintenanceTaskTestBase {
// Branch should be compacted from 2 files to 1
assertThat(
- table.snapshot(branchName).dataManifests(table.io()).stream()
+ table.snapshot(TEST_BRANCH).dataManifests(table.io()).stream()
.flatMap(
m ->
StreamSupport.stream(
@@ -646,7 +673,7 @@ class TestRewriteDataFiles extends MaintenanceTaskTestBase {
.count())
.isEqualTo(1);
SimpleDataUtil.assertTableRecords(
- table, ImmutableList.of(createRecord(1, "a"), createRecord(2, "b")),
branchName);
+ table, ImmutableList.of(createRecord(1, "a"), createRecord(2, "b")),
TEST_BRANCH);
// Main should be untouched with 3 files
assertFileNum(table, 3, 0);
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFilesConfig.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFilesConfig.java
index 665a82ea15..73cfd4eb17 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFilesConfig.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestRewriteDataFilesConfig.java
@@ -50,6 +50,7 @@ public class TestRewriteDataFilesConfig extends
OperatorTestBase {
input.put(RewriteDataFilesConfig.SCHEDULE_ON_DATA_FILE_COUNT, "20");
input.put(RewriteDataFilesConfig.SCHEDULE_ON_DATA_FILE_SIZE, "30");
input.put(RewriteDataFilesConfig.SCHEDULE_ON_INTERVAL_SECOND, "60");
+ input.put(RewriteDataFilesConfig.BRANCH, "custom");
input.put("other.config", "should-be-ignored");
}
@@ -100,6 +101,7 @@ public class TestRewriteDataFilesConfig extends
OperatorTestBase {
assertThat(config.partialProgressEnable()).isTrue();
assertThat(config.partialProgressMaxCommits()).isEqualTo(5);
assertThat(config.maxRewriteBytes()).isEqualTo(1024L);
+ assertThat(config.branch()).isEqualTo("custom");
// check the config about the schedule
assertThat(config.scheduleOnCommitCount()).isEqualTo(10);
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
index 049589f3a4..af21f34937 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSinkTableMaintenance.java
@@ -42,6 +42,7 @@ import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.actions.SizeBasedFileRewritePlanner;
@@ -50,10 +51,13 @@ import org.apache.iceberg.flink.FlinkWriteOptions;
import org.apache.iceberg.flink.MiniFlinkClusterExtension;
import org.apache.iceberg.flink.SimpleDataUtil;
import org.apache.iceberg.flink.TestFixtures;
+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.FlinkMaintenanceConfig;
import org.apache.iceberg.flink.maintenance.api.LockConfig;
import org.apache.iceberg.flink.maintenance.api.RewriteDataFilesConfig;
+import org.apache.iceberg.flink.maintenance.operator.ManualSource;
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;
@@ -61,6 +65,7 @@ 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.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.FieldSource;
@@ -436,6 +441,87 @@ class TestIcebergSinkTableMaintenance extends
TestFlinkIcebergSinkBase {
}
}
+ // TODO: Also run with the JDBC lock once JdbcLockFactory creates the lock
table safely. The
+ // trigger manager and the lock remover both create it on open, and the
second CREATE TABLE
+ // fails with "table already exists" when they race.
+ @Test
+ public void testConvertEqualityDeletesWithRewriteEnabledE2e() throws
Exception {
+ setupLockConfig("");
+ setupCompactionConfig();
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+ flinkConf.put(FlinkWriteOptions.BRANCH.key(), "staging");
+ flinkConf.put(ConvertEqualityDeletesConfig.TARGET_BRANCH,
SnapshotRef.MAIN_BRANCH);
+ flinkConf.put(FlinkMaintenanceConfig.RATE_LIMIT, "1");
+ flinkConf.put(FlinkMaintenanceConfig.LOCK_CHECK_DELAY, "1");
+
+ new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
+ .appendToTable(ImmutableList.of(SimpleDataUtil.createRecord(1,
"aaa")));
+
+ ManualSource<Row> source = new ManualSource<>(env, ROW_TYPE_INFO);
+ DataStream<RowData> dataStream =
+ source
+ .dataStream()
+ .map(CONVERTER::toInternal,
FlinkCompatibilityUtil.toTypeInfo(SimpleDataUtil.ROW_TYPE));
+
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append();
+
+ source.sendRecord(Row.of(1, "bbb"));
+ JobClient jobClient = env.executeAsync("Test Rewrite On Convert Target
Branch E2E");
+ try {
+ Awaitility.await()
+ .atMost(Duration.ofMinutes(2))
+ .pollInterval(Duration.ofMillis(200))
+ .untilAsserted(
+ () -> {
+ assertThat(dvCountOnMain(table)).isEqualTo(0);
+ assertThat(getDataFiles(table.currentSnapshot(),
table)).hasSize(1);
+ SimpleDataUtil.assertTableRecords(
+ table, ImmutableList.of(SimpleDataUtil.createRecord(1,
"bbb")));
+ });
+ } finally {
+ jobClient.cancel();
+ }
+ }
+
+ @Test
+ void testRewriteBranchMustMatchConvertTargetBranch() {
+ setupLockConfig("");
+ setupCompactionConfig();
+ setupConvertEqualityDeletesConfig();
+ upgradeToFormatV3();
+ flinkConf.put(FlinkWriteOptions.BRANCH.key(), "staging");
+ flinkConf.put(ConvertEqualityDeletesConfig.TARGET_BRANCH, "main");
+ flinkConf.put(RewriteDataFilesConfig.BRANCH, "staging");
+
+ 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));
+
+ assertThatThrownBy(
+ () ->
+ IcebergSink.forRowData(dataStream)
+ .table(table)
+ .tableLoader(tableLoader)
+ .upsert(true)
+ .equalityFieldColumns(Lists.newArrayList("id"))
+ .setAll(flinkConf)
+ .append())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Invalid rewrite branch staging. Expected rewrite branch main, "
+ + "the convert equality deletes target branch. Unset config "
+ + RewriteDataFilesConfig.BRANCH
+ + " to default to it.");
+ }
+
private static long dvCountOnMain(Table table) throws IOException {
table.refresh();
if (table.currentSnapshot() == null) {
