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 2026/04/17 17:58:14 UTC
(iceberg) branch main updated: Flink: Set generator parallelism to match input in DynamicIcebergSink (#15849)
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 8f1f483985 Flink: Set generator parallelism to match input in DynamicIcebergSink (#15849)
8f1f483985 is described below
commit 8f1f483985bfbef3c19472b9e2ec474593fee219
Author: Sachin Ranjalkar <52...@users.noreply.github.com>
AuthorDate: Fri Apr 17 23:28:06 2026 +0530
Flink: Set generator parallelism to match input in DynamicIcebergSink (#15849)
---
.../flink/sink/dynamic/DynamicIcebergSink.java | 1 +
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 27 ++++++++++++++++++++++
.../flink/sink/dynamic/DynamicIcebergSink.java | 1 +
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 27 ++++++++++++++++++++++
.../flink/sink/dynamic/DynamicIcebergSink.java | 1 +
.../flink/sink/dynamic/TestDynamicIcebergSink.java | 27 ++++++++++++++++++++++
6 files changed, 84 insertions(+)
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index 218fa2d911..63f5e61911 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -408,6 +408,7 @@ public class DynamicIcebergSink
tableCreator,
caseSensitive,
dropUnusedColumns))
+ .setParallelism(input.getParallelism())
.uid(prefixIfNotNull(uidPrefix, "-generator"))
.name(operatorName("generator"))
.returns(type);
diff --git a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 27b1e3d84a..aa92ae8ceb 100644
--- a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++ b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -1196,6 +1196,33 @@ class TestDynamicIcebergSink extends TestFlinkIcebergSinkBase {
"Sink Committer: --sink");
}
+ @Test
+ void testGeneratorDefaultParallelism() {
+ StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ streamEnv.setParallelism(4);
+
+ DataStreamSource<DynamicIcebergDataImpl> source =
+ streamEnv.fromData(Collections.emptySet(), TypeInformation.of(new TypeHint<>() {}));
+ source.setParallelism(8);
+
+ DynamicIcebergSink.forInput(source)
+ .generator(new Generator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .uidPrefix("test")
+ .append();
+
+ // Since the generator parallelism is not directly accessible via the returned DataStreamSink,
+ // inspect the stream graph to verify the generator inherits the input source parallelism.
+ int generatorParallelism =
+ streamEnv.getStreamGraph().getStreamNodes().stream()
+ .filter(node -> "test--generator".equals(node.getTransformationUID()))
+ .findFirst()
+ .map(StreamNode::getParallelism)
+ .orElseThrow(() -> new AssertionError("Generator node not found"));
+
+ assertThat(generatorParallelism).isEqualTo(source.getParallelism());
+ }
+
private Set<String> createSinkAndReturnUIds(String uidPrefix) {
StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index 4b5c9bef41..a00eba492a 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -407,6 +407,7 @@ public class DynamicIcebergSink
tableCreator,
caseSensitive,
dropUnusedColumns))
+ .setParallelism(input.getParallelism())
.uid(prefixIfNotNull(uidPrefix, "-generator"))
.name(operatorName("generator"))
.returns(type);
diff --git a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 27b1e3d84a..aa92ae8ceb 100644
--- a/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++ b/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -1196,6 +1196,33 @@ class TestDynamicIcebergSink extends TestFlinkIcebergSinkBase {
"Sink Committer: --sink");
}
+ @Test
+ void testGeneratorDefaultParallelism() {
+ StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ streamEnv.setParallelism(4);
+
+ DataStreamSource<DynamicIcebergDataImpl> source =
+ streamEnv.fromData(Collections.emptySet(), TypeInformation.of(new TypeHint<>() {}));
+ source.setParallelism(8);
+
+ DynamicIcebergSink.forInput(source)
+ .generator(new Generator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .uidPrefix("test")
+ .append();
+
+ // Since the generator parallelism is not directly accessible via the returned DataStreamSink,
+ // inspect the stream graph to verify the generator inherits the input source parallelism.
+ int generatorParallelism =
+ streamEnv.getStreamGraph().getStreamNodes().stream()
+ .filter(node -> "test--generator".equals(node.getTransformationUID()))
+ .findFirst()
+ .map(StreamNode::getParallelism)
+ .orElseThrow(() -> new AssertionError("Generator node not found"));
+
+ assertThat(generatorParallelism).isEqualTo(source.getParallelism());
+ }
+
private Set<String> createSinkAndReturnUIds(String uidPrefix) {
StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
index 6f5fb945a1..9c8a4d156f 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/DynamicIcebergSink.java
@@ -502,6 +502,7 @@ public class DynamicIcebergSink
tableCreator,
caseSensitive,
dropUnusedColumns))
+ .setParallelism(input.getParallelism())
.uid(prefixIfNotNull(uidPrefix, "-generator"))
.name(operatorName("generator"))
.returns(type);
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
index 2a46f8021c..4e75115010 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestDynamicIcebergSink.java
@@ -1342,6 +1342,33 @@ class TestDynamicIcebergSink extends TestFlinkIcebergSinkBase {
"Sink Committer: --sink");
}
+ @Test
+ void testGeneratorDefaultParallelism() {
+ StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ streamEnv.setParallelism(4);
+
+ DataStreamSource<DynamicIcebergDataImpl> source =
+ streamEnv.fromData(Collections.emptySet(), TypeInformation.of(new TypeHint<>() {}));
+ source.setParallelism(8);
+
+ DynamicIcebergSink.forInput(source)
+ .generator(new Generator())
+ .catalogLoader(CATALOG_EXTENSION.catalogLoader())
+ .uidPrefix("test")
+ .append();
+
+ // Since the generator parallelism is not directly accessible via the returned DataStreamSink,
+ // inspect the stream graph to verify the generator inherits the input source parallelism.
+ int generatorParallelism =
+ streamEnv.getStreamGraph().getStreamNodes().stream()
+ .filter(node -> "test--generator".equals(node.getTransformationUID()))
+ .findFirst()
+ .map(StreamNode::getParallelism)
+ .orElseThrow(() -> new AssertionError("Generator node not found"));
+
+ assertThat(generatorParallelism).isEqualTo(source.getParallelism());
+ }
+
private Set<String> createSinkAndReturnUIds(String uidPrefix) {
StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();