You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2025/07/24 07:28:55 UTC
(iceberg) 03/03: Flink: backport PR #13646 to 1.19 for fail compaction on V3 tables
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
commit 53ba1cfbade47de21fda777bffc5c0e7360919d7
Author: Steven Wu <st...@gmail.com>
AuthorDate: Wed Jul 23 14:26:05 2025 -0700
Flink: backport PR #13646 to 1.19 for fail compaction on V3 tables
---
.../flink/actions/RewriteDataFilesAction.java | 4 +++
.../operator/DataFileRewritePlanner.java | 6 ++++
.../flink/actions/TestRewriteDataFilesAction.java | 34 ++++++++++++++++++++++
.../maintenance/operator/OperatorTestBase.java | 14 +++++++--
.../operator/TestDataFileRewritePlanner.java | 14 +++++++++
5 files changed, 70 insertions(+), 2 deletions(-)
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/actions/RewriteDataFilesAction.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/actions/RewriteDataFilesAction.java
index 670abebcb5..4cf30ed904 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/actions/RewriteDataFilesAction.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/actions/RewriteDataFilesAction.java
@@ -24,6 +24,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.iceberg.CombinedScanTask;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
import org.apache.iceberg.actions.BaseRewriteDataFilesAction;
import org.apache.iceberg.flink.source.RowDataRewriter;
import org.apache.iceberg.io.FileIO;
@@ -38,6 +39,9 @@ public class RewriteDataFilesAction extends BaseRewriteDataFilesAction<RewriteDa
super(table);
this.env = env;
this.maxParallelism = env.getParallelism();
+ Preconditions.checkArgument(
+ !TableUtil.supportsRowLineage(table),
+ "Flink does not support compaction on row lineage enabled tables (V3+)");
}
@Override
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
index e36c9c6610..eefd5708fa 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/DataFileRewritePlanner.java
@@ -29,6 +29,8 @@ import org.apache.flink.util.Collector;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.SerializableTable;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
import org.apache.iceberg.actions.BinPackRewriteFilePlanner;
import org.apache.iceberg.actions.FileRewritePlan;
import org.apache.iceberg.actions.RewriteDataFiles;
@@ -86,6 +88,10 @@ public class DataFileRewritePlanner
@Override
public void open(Configuration parameters) throws Exception {
tableLoader.open();
+ Table table = tableLoader.loadTable();
+ Preconditions.checkArgument(
+ !TableUtil.supportsRowLineage(table),
+ "Flink does not support compaction on row lineage enabled tables (V3+)");
this.errorCounter =
TableMaintenanceMetrics.groupFor(getRuntimeContext(), tableName, taskName, taskIndex)
.counter(TableMaintenanceMetrics.ERROR_COUNTER);
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
index bb33983b32..0117197802 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/actions/TestRewriteDataFilesAction.java
@@ -65,6 +65,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.util.Pair;
+import org.assertj.core.api.Assumptions;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
@@ -147,8 +148,23 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
super.clean();
}
+ @TestTemplate
+ public void testFailureOnV3Table() {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isGreaterThanOrEqualTo(3);
+
+ assertThatThrownBy(
+ () -> Actions.forTable(icebergTableUnPartitioned).rewriteDataFiles().execute())
+ .hasMessageContaining(
+ "Flink does not support compaction on row lineage enabled tables (V3+)")
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
@TestTemplate
public void testRewriteDataFilesEmptyTable() throws Exception {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
assertThat(icebergTableUnPartitioned.currentSnapshot()).isNull();
Actions.forTable(icebergTableUnPartitioned).rewriteDataFiles().execute();
assertThat(icebergTableUnPartitioned.currentSnapshot()).isNull();
@@ -156,6 +172,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
@TestTemplate
public void testRewriteDataFilesUnpartitionedTable() throws Exception {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
sql("INSERT INTO %s SELECT 1, 'hello'", TABLE_NAME_UNPARTITIONED);
sql("INSERT INTO %s SELECT 2, 'world'", TABLE_NAME_UNPARTITIONED);
@@ -186,6 +205,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
@TestTemplate
public void testRewriteDataFilesPartitionedTable() throws Exception {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
sql("INSERT INTO %s SELECT 1, 'hello' ,'a'", TABLE_NAME_PARTITIONED);
sql("INSERT INTO %s SELECT 2, 'hello' ,'a'", TABLE_NAME_PARTITIONED);
sql("INSERT INTO %s SELECT 3, 'world' ,'b'", TABLE_NAME_PARTITIONED);
@@ -228,6 +250,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
@TestTemplate
public void testRewriteDataFilesWithFilter() throws Exception {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
sql("INSERT INTO %s SELECT 1, 'hello' ,'a'", TABLE_NAME_PARTITIONED);
sql("INSERT INTO %s SELECT 2, 'hello' ,'a'", TABLE_NAME_PARTITIONED);
sql("INSERT INTO %s SELECT 3, 'world' ,'a'", TABLE_NAME_PARTITIONED);
@@ -275,6 +300,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
@TestTemplate
public void testRewriteLargeTableHasResiduals() throws IOException {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
// all records belong to the same partition
List<String> records1 = Lists.newArrayList();
List<String> records2 = Lists.newArrayList();
@@ -337,6 +365,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
*/
@TestTemplate
public void testRewriteAvoidRepeateCompress() throws IOException {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
List<Record> expected = Lists.newArrayList();
Schema schema = icebergTableUnPartitioned.schema();
GenericAppenderFactory genericAppenderFactory = new GenericAppenderFactory(schema);
@@ -401,6 +432,9 @@ public class TestRewriteDataFilesAction extends CatalogTestBase {
@TestTemplate
public void testRewriteNoConflictWithEqualityDeletes() throws IOException {
+ // Flink does not support compaction on row lineage enabled tables (V3+)
+ Assumptions.assumeThat(formatVersion).isLessThan(3);
+
// Add 2 data files
sql("INSERT INTO %s SELECT 1, 'hello'", TABLE_NAME_WITH_PK);
sql("INSERT INTO %s SELECT 2, 'world'", TABLE_NAME_WITH_PK);
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
index c690bdec26..b1a2b710fa 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
@@ -44,6 +44,7 @@ import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TestHelpers;
import org.apache.iceberg.data.FileHelpers;
import org.apache.iceberg.data.GenericAppenderHelper;
@@ -117,6 +118,11 @@ public class OperatorTestBase {
}
protected static Table createTable() {
+ // only test V2 tables as compaction doesn't support V3 with row lineage
+ return createTable("2");
+ }
+
+ protected static Table createTable(String formatVersion) {
return CATALOG_EXTENSION
.catalog()
.createTable(
@@ -124,7 +130,11 @@ public class OperatorTestBase {
SimpleDataUtil.SCHEMA,
PartitionSpec.unpartitioned(),
null,
- ImmutableMap.of("flink.max-continuous-empty-commits", "100000"));
+ ImmutableMap.of(
+ TableProperties.FORMAT_VERSION,
+ formatVersion,
+ "flink.max-continuous-empty-commits",
+ "100000"));
}
protected static Table createTableWithDelete() {
@@ -146,7 +156,7 @@ public class OperatorTestBase {
SimpleDataUtil.SCHEMA,
PartitionSpec.builderFor(SimpleDataUtil.SCHEMA).identity("data").build(),
null,
- ImmutableMap.of("flink.max-continuous-empty-commits", "100000"));
+ ImmutableMap.of("format-version", "2", "flink.max-continuous-empty-commits", "100000"));
}
protected void insert(Table table, Integer id, String data) throws IOException {
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
index 1d7e29813a..8798ff23a2 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestDataFileRewritePlanner.java
@@ -22,6 +22,7 @@ import static org.apache.iceberg.actions.SizeBasedFileRewritePlanner.MIN_INPUT_F
import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.newDataFiles;
import static org.apache.iceberg.flink.maintenance.operator.RewriteUtil.planDataFileRewrite;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.List;
import java.util.Set;
@@ -39,6 +40,19 @@ import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.junit.jupiter.api.Test;
class TestDataFileRewritePlanner extends OperatorTestBase {
+ @Test
+ void testFailsOnV3Table() throws Exception {
+ Table table = createTable("3");
+ Set<DataFile> expected = Sets.newHashSetWithExpectedSize(3);
+ insert(table, 1, "a");
+ expected.addAll(newDataFiles(table));
+
+ assertThatThrownBy(() -> planDataFileRewrite(tableLoader()))
+ .hasMessageContaining(
+ "Flink does not support compaction on row lineage enabled tables (V3+)")
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
@Test
void testUnpartitioned() throws Exception {
Set<DataFile> expected = Sets.newHashSetWithExpectedSize(3);