You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2025/09/28 10:28:45 UTC
(iceberg) branch main updated: Flink: Backport add uid-suffix write option to prevent operator UID hash collisions (#14193)
This is an automated email from the ASF dual-hosted git repository.
pvary 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 69b5caaa97 Flink: Backport add uid-suffix write option to prevent operator UID hash collisions (#14193)
69b5caaa97 is described below
commit 69b5caaa9776f3dbb8903a6ce358c74485bf8d11
Author: Rodrigo <rm...@apple.com>
AuthorDate: Sun Sep 28 03:28:40 2025 -0700
Flink: Backport add uid-suffix write option to prevent operator UID hash collisions (#14193)
Backports #14063
---
.../org/apache/iceberg/flink/FlinkWriteConf.java | 8 ++++
.../apache/iceberg/flink/FlinkWriteOptions.java | 4 ++
.../org/apache/iceberg/flink/sink/IcebergSink.java | 8 ++--
.../apache/iceberg/flink/TestFlinkTableSink.java | 55 ++++++++++++++++++++++
.../org/apache/iceberg/flink/FlinkWriteConf.java | 8 ++++
.../apache/iceberg/flink/FlinkWriteOptions.java | 4 ++
.../org/apache/iceberg/flink/sink/IcebergSink.java | 8 ++--
.../apache/iceberg/flink/TestFlinkTableSink.java | 55 ++++++++++++++++++++++
8 files changed, 142 insertions(+), 8 deletions(-)
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 222a1e8104..66fd098077 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
@@ -208,6 +208,14 @@ public class FlinkWriteConf {
.parse();
}
+ public String uidSuffix() {
+ return confParser
+ .stringConf()
+ .option(FlinkWriteOptions.UID_SUFFIX.key())
+ .defaultValue(FlinkWriteOptions.UID_SUFFIX.defaultValue())
+ .parse();
+ }
+
public Integer writeParallelism() {
return confParser.intConf().option(FlinkWriteOptions.WRITE_PARALLELISM.key()).parseOptional();
}
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 6bdb01c3f5..e68e64ac57 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
@@ -87,4 +87,8 @@ public class FlinkWriteOptions {
@Experimental
public static final ConfigOption<Duration> TABLE_REFRESH_INTERVAL =
ConfigOptions.key("table-refresh-interval").durationType().noDefaultValue();
+
+ // specify the uidSuffix to be used for the underlying IcebergSink
+ public static final ConfigOption<String> UID_SUFFIX =
+ ConfigOptions.key("uid-suffix").stringType().defaultValue("");
}
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 f2d3e1fe44..f27db4f2b4 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
@@ -320,7 +320,6 @@ public class IcebergSink
public static class Builder implements IcebergSinkBuilder<Builder> {
private TableLoader tableLoader;
- private String uidSuffix = "";
private Function<String, DataStream<RowData>> inputCreator = null;
@Deprecated private TableSchema tableSchema;
private ResolvedSchema resolvedSchema;
@@ -596,7 +595,7 @@ public class IcebergSink
* @return {@link Builder} to connect the iceberg table.
*/
public Builder uidSuffix(String newSuffix) {
- this.uidSuffix = newSuffix;
+ writeOptions.put(FlinkWriteOptions.UID_SUFFIX.key(), newSuffix);
return this;
}
@@ -665,11 +664,12 @@ public class IcebergSink
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+
return new IcebergSink(
tableLoader,
table,
snapshotSummary,
- uidSuffix,
+ flinkWriteConf.uidSuffix(),
SinkUtil.writeProperties(flinkWriteConf.dataFileFormat(), flinkWriteConf, table),
resolvedSchema != null
? toFlinkRowType(table.schema(), resolvedSchema)
@@ -690,7 +690,7 @@ public class IcebergSink
@Override
public DataStreamSink<RowData> append() {
IcebergSink sink = build();
- String suffix = defaultSuffix(uidSuffix, table.name());
+ String suffix = defaultSuffix(sink.uidSuffix, table.name());
DataStream<RowData> rowDataInput = inputCreator.apply(suffix);
// Please note that V2 sink framework will apply the uid here to the framework created
// operators like writer,
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
index d99f657a11..572ff68202 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
@@ -263,4 +263,59 @@ public class TestFlinkTableSink extends CatalogTestBase {
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
}
}
+
+ // Test multiple IcebergSink instances writing to the same table in separate DAG branches.
+ // This ensures the v2 sink can handle multiple sink operators for the same table
+ // without naming collisions or operator conflicts when using statement set execution.
+ @TestTemplate
+ public void testIcebergSinkDifferentDAG() throws Exception {
+ assumeThat(useV2Sink).isTrue();
+
+ // Disable sink reuse optimization to force creation of two separate IcebergSink instances
+ getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .setString("table.optimizer.reuse-sink-enabled", "false");
+
+ // Register the rows into a temporary table.
+ getTableEnv()
+ .createTemporaryView(
+ "sourceTable",
+ getTableEnv()
+ .fromValues(
+ SimpleDataUtil.FLINK_SCHEMA.toSourceRowDataType(),
+ Expressions.row(1, "hello"),
+ Expressions.row(2, "world"),
+ Expressions.row(3, (String) null),
+ Expressions.row(null, "bar")));
+
+ getTableEnv()
+ .createTemporaryView(
+ "sourceTable1",
+ getTableEnv()
+ .fromValues(
+ SimpleDataUtil.FLINK_SCHEMA.toSourceRowDataType(),
+ Expressions.row(1, "hello"),
+ Expressions.row(2, "world"),
+ Expressions.row(3, (String) null),
+ Expressions.row(null, "bar")));
+
+ // Redirect the records from source table to destination table.
+ sql(
+ "EXECUTE STATEMENT SET\n"
+ + "BEGIN\n"
+ + "INSERT INTO %s /*+ OPTIONS('uid-suffix'='source1') */ SELECT id,data from sourceTable;\n"
+ + "INSERT INTO %s /*+ OPTIONS('uid-suffix'='source2') */ SELECT id,data from sourceTable1;\n"
+ + "END;",
+ TABLE_NAME, TABLE_NAME);
+
+ // Assert the table records as expected.
+ SimpleDataUtil.assertTableRecords(
+ icebergTable,
+ Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "hello"),
+ SimpleDataUtil.createRecord(2, "world"),
+ SimpleDataUtil.createRecord(3, null),
+ SimpleDataUtil.createRecord(null, "bar")));
+ }
}
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 222a1e8104..66fd098077 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
@@ -208,6 +208,14 @@ public class FlinkWriteConf {
.parse();
}
+ public String uidSuffix() {
+ return confParser
+ .stringConf()
+ .option(FlinkWriteOptions.UID_SUFFIX.key())
+ .defaultValue(FlinkWriteOptions.UID_SUFFIX.defaultValue())
+ .parse();
+ }
+
public Integer writeParallelism() {
return confParser.intConf().option(FlinkWriteOptions.WRITE_PARALLELISM.key()).parseOptional();
}
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 6bdb01c3f5..e68e64ac57 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
@@ -87,4 +87,8 @@ public class FlinkWriteOptions {
@Experimental
public static final ConfigOption<Duration> TABLE_REFRESH_INTERVAL =
ConfigOptions.key("table-refresh-interval").durationType().noDefaultValue();
+
+ // specify the uidSuffix to be used for the underlying IcebergSink
+ public static final ConfigOption<String> UID_SUFFIX =
+ ConfigOptions.key("uid-suffix").stringType().defaultValue("");
}
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 752882a9d6..593230262f 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
@@ -321,7 +321,6 @@ public class IcebergSink
public static class Builder implements IcebergSinkBuilder<Builder> {
private TableLoader tableLoader;
- private String uidSuffix = "";
private Function<String, DataStream<RowData>> inputCreator = null;
@Deprecated private TableSchema tableSchema;
private ResolvedSchema resolvedSchema;
@@ -597,7 +596,7 @@ public class IcebergSink
* @return {@link Builder} to connect the iceberg table.
*/
public Builder uidSuffix(String newSuffix) {
- this.uidSuffix = newSuffix;
+ writeOptions.put(FlinkWriteOptions.UID_SUFFIX.key(), newSuffix);
return this;
}
@@ -666,11 +665,12 @@ public class IcebergSink
FlinkMaintenanceConfig flinkMaintenanceConfig =
new FlinkMaintenanceConfig(table, writeOptions, readableConfig);
+
return new IcebergSink(
tableLoader,
table,
snapshotSummary,
- uidSuffix,
+ flinkWriteConf.uidSuffix(),
SinkUtil.writeProperties(flinkWriteConf.dataFileFormat(), flinkWriteConf, table),
resolvedSchema != null
? toFlinkRowType(table.schema(), resolvedSchema)
@@ -691,7 +691,7 @@ public class IcebergSink
@Override
public DataStreamSink<RowData> append() {
IcebergSink sink = build();
- String suffix = defaultSuffix(uidSuffix, table.name());
+ String suffix = defaultSuffix(sink.uidSuffix, table.name());
DataStream<RowData> rowDataInput = inputCreator.apply(suffix);
// Please note that V2 sink framework will apply the uid here to the framework created
// operators like writer,
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
index d99f657a11..572ff68202 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
@@ -263,4 +263,59 @@ public class TestFlinkTableSink extends CatalogTestBase {
sql("DROP TABLE IF EXISTS %s.%s", flinkDatabase, tableName);
}
}
+
+ // Test multiple IcebergSink instances writing to the same table in separate DAG branches.
+ // This ensures the v2 sink can handle multiple sink operators for the same table
+ // without naming collisions or operator conflicts when using statement set execution.
+ @TestTemplate
+ public void testIcebergSinkDifferentDAG() throws Exception {
+ assumeThat(useV2Sink).isTrue();
+
+ // Disable sink reuse optimization to force creation of two separate IcebergSink instances
+ getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .setString("table.optimizer.reuse-sink-enabled", "false");
+
+ // Register the rows into a temporary table.
+ getTableEnv()
+ .createTemporaryView(
+ "sourceTable",
+ getTableEnv()
+ .fromValues(
+ SimpleDataUtil.FLINK_SCHEMA.toSourceRowDataType(),
+ Expressions.row(1, "hello"),
+ Expressions.row(2, "world"),
+ Expressions.row(3, (String) null),
+ Expressions.row(null, "bar")));
+
+ getTableEnv()
+ .createTemporaryView(
+ "sourceTable1",
+ getTableEnv()
+ .fromValues(
+ SimpleDataUtil.FLINK_SCHEMA.toSourceRowDataType(),
+ Expressions.row(1, "hello"),
+ Expressions.row(2, "world"),
+ Expressions.row(3, (String) null),
+ Expressions.row(null, "bar")));
+
+ // Redirect the records from source table to destination table.
+ sql(
+ "EXECUTE STATEMENT SET\n"
+ + "BEGIN\n"
+ + "INSERT INTO %s /*+ OPTIONS('uid-suffix'='source1') */ SELECT id,data from sourceTable;\n"
+ + "INSERT INTO %s /*+ OPTIONS('uid-suffix'='source2') */ SELECT id,data from sourceTable1;\n"
+ + "END;",
+ TABLE_NAME, TABLE_NAME);
+
+ // Assert the table records as expected.
+ SimpleDataUtil.assertTableRecords(
+ icebergTable,
+ Lists.newArrayList(
+ SimpleDataUtil.createRecord(1, "hello"),
+ SimpleDataUtil.createRecord(2, "world"),
+ SimpleDataUtil.createRecord(3, null),
+ SimpleDataUtil.createRecord(null, "bar")));
+ }
}