Flink: SQL support for dynamic iceberg sink by swapna267 · Pull Request #15279 · apache/iceberg

swapna267 · 2026-02-09T18:37:24Z

This PR introduces a SQL table connector for using the dynamic iceberg sink.

Two new configuration options have been added to FlinkCreateTableOptions:

use-dynamic-iceberg-sink (boolean): Enable/disable dynamic sink functionality
dynamic-record-generator-impl (string): Fully qualified class name of the DynamicTableRecordGenerator implementation

Example SQL,

  CREATE TABLE dynamic_sink_table (
      id BIGINT,
      data STRING,
      database_name STRING,
      table_name STRING
  ) WITH (
      'connector' = 'iceberg',
      'catalog-type' = 'hadoop',
      'catalog-name' = 'my_catalog',
      'warehouse' = 's3://my-warehouse/',
      'use-dynamic-iceberg-sink' = 'true',
      'dynamic-record-generator-impl' = 'com.example.MyDynamicRecordGenerator',
      'table.props.write.format.default' = 'parquet',
      'table.props.write.target-file-size-bytes' = '134217728'
  );

  -- Insert data that will be routed to different tables based on database_name and table_name
  INSERT INTO dynamic_sink_table VALUES
      (1, 'record1', 'sales', 'orders'),
      (2, 'record2', 'sales', 'customers'),
      (3, 'record3', 'inventory', 'products');

Planning to provide a CustomVariantToDynamicRecordGenerator that can handle Flink VARIANT type column to generate records of different schemas landing in tables of corresponding schema.
Will add that in a different PR.

mxm

Thanks @swapna267! This looks great.

mxm · 2026-02-10T15:11:46Z

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java

  }
+
+  @TestTemplate
+  public void testCreateDynamicIcebergSink() throws DatabaseAlreadyExistException {


Could we verify this test works with both the old FlinkSink and the new IcebergSink?

This test in particular is testing the DynamicIcebergSink only by setting use-dynamic-iceberg-sink to true.

But i also see, TestIcebergConnector is not testing the new IcebergSink code path. Partially it's covered in TestFlinkTableSink (where iceberg tables are created in Iceberg catalog).

If my understanding is right, i prefer to put that into separate PR.

That makes sense. The test is fine as-is.

mxm · 2026-02-10T15:12:26Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

+    String dynamicRecordGeneratorImpl =
+        flinkConf.get(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+    Preconditions.checkNotNull(
+        dynamicRecordGeneratorImpl,
+        "%s must be specified when use-dynamic-iceberg-sink is true",
+        FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL.key());


Should we add a test to verify these conditions?

Sure can add. Don't see such detailed one's in general . Also concerned about the time tests take today to complete.

We have many such tests for Dynamic Sink. Not specifying the record generator will probably error when it's being created, but it would still be nice to check for the particular error message reported back to the user. I'll leave it up to you to add it or not.

Guosmilesmile

Thanks for the Pr！Left some comments.

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java

Guosmilesmile · 2026-02-11T13:57:52Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java

+
+  private TableCreator createTableCreator() {
+    final Map<String, String> tableProperties =
+        org.apache.iceberg.util.PropertyUtil.propertiesWithPrefix(writeProps, "table.props.");


If I’m not mistaken, if we want to set the table property write.parquet.row-group-size-bytes, do we need to specify it here as table.props.write.parquet.row-group-size-bytes? I think this should be documented and we should add a corresponding test case.

Yes right. When doing CREATE TABLE in flink catalog , we pass in catalog configuration here.
table.props prefix is used to separate out the physical Iceberg table properties.

Basic documentation about the connector is here,
https://iceberg.apache.org/docs/nightly/flink-connector/
Once we have all functionality (dynamic record generator impl is coming in next PR), will add details there.

I combined this in existing test case , https://github.com/swapna267/iceberg/blob/bd2d500f07fb24d05111b6dabc9a8e77637a922c/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java#L393
I can pull it out into another one if we think it's required.

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

swapna267 · 2026-02-11T20:54:42Z

Thanks @mxm and @Guosmilesmile for the review. Replied on some comments.

mxm · 2026-02-12T10:58:10Z

flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestIcebergConnector.java

+            .getCatalogLoader()
+            .loadCatalog()
+            .loadTable(TableIdentifier.of(databaseName(), tableName()));
+    assertThat(table.properties()).containsEntry("key1", "val1");


Could we also verify the records written to the table?

mxm · 2026-02-12T11:01:28Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

+    String dynamicRecordGeneratorImpl =
+        flinkConf.get(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+    Preconditions.checkNotNull(
+        dynamicRecordGeneratorImpl,
+        "%s must be specified when use-dynamic-iceberg-sink is true",
+        FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL.key());


We have many such tests for Dynamic Sink. Not specifying the record generator will probably error when it's being created, but it would still be nice to check for the particular error message reported back to the user. I'll leave it up to you to add it or not.

mxm · 2026-02-12T11:03:30Z

LGTM, just some minor comments.

pvary · 2026-02-17T10:43:24Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

-        tableLoader, resolvedSchema, context.getConfiguration(), writeProps);
+      return new IcebergTableSink(
+          tableLoader,
+          resolvedCatalogTable.getResolvedSchema(),


Why did we remove the filtering for the physical columns?

pvary · 2026-02-17T10:46:37Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

+      Context context, Configuration flinkConf, Map<String, String> writeProps) {
+    String dynamicRecordGeneratorImpl =
+        flinkConf.get(FlinkCreateTableOptions.DYNAMIC_RECORD_GENERATOR_IMPL);
+    Preconditions.checkNotNull(


I have received several comments that instead of checkNotNull we should use checkArgument and a message like Invalid dynamic record generator value: %s. %s must be specified when use-dynamic-iceberg-sink is true.

pvary · 2026-02-17T10:47:26Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

+    return new IcebergTableSink(
+        catalogLoader,
+        dynamicRecordGeneratorImpl,
+        resolvedCatalogTable.getResolvedSchema(),


Do we need something like this?

ResolvedSchema resolvedSchema = ResolvedSchema.of( resolvedCatalogTable.getResolvedSchema().getColumns().stream() .filter(Column::isPhysical) .collect(Collectors.toList()));

pvary · 2026-02-17T10:47:43Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

+
+  private static FlinkCatalog createCatalogLoader(
+      Map<String, String> tableProps, String catalogName) {
+    Preconditions.checkNotNull(


use checkArgument and "standard error message"

I see that this is only a move for this check. Do you think it would cause any issues if we change this to the new standard?

pvary · 2026-02-17T10:48:51Z

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

+    org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();
+    FlinkCatalogFactory factory = new FlinkCatalogFactory();
+    return (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);


nit: Just to get rid of the org.apache.hadoop.conf.Configuration

Suggested change

org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();

FlinkCatalogFactory factory = new FlinkCatalogFactory();

return (FlinkCatalog) factory.createCatalog(catalogName, tableProps, hadoopConf);

FlinkCatalogFactory factory = new FlinkCatalogFactory();

return (FlinkCatalog) factory.createCatalog(catalogName, tableProps, FlinkCatalogFactory.clusterHadoopConf(););

pvary · 2026-02-17T10:55:17Z

Nice stuff @swapna267!

Could you please update the documentation too?

sql support for dynamic iceberg sink

845d91a

github-actions bot added the flink label Feb 9, 2026

swapna267 added 2 commits February 9, 2026 11:31

fix the comment

45c1f06

fix test

bd2d500

mxm reviewed Feb 10, 2026

View reviewed changes

Guosmilesmile reviewed Feb 11, 2026

View reviewed changes

pr review comments

17f2f1a

mxm reviewed Feb 12, 2026

View reviewed changes

swapna267 added 2 commits February 12, 2026 11:26

enhance tests

62b8e55

fix test

2638c69

mxm approved these changes Feb 13, 2026

View reviewed changes

Guosmilesmile approved these changes Feb 16, 2026

View reviewed changes

pvary reviewed Feb 17, 2026

View reviewed changes

Conversation

swapna267 commented Feb 9, 2026

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

swapna267 Feb 11, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Guosmilesmile left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

swapna267 commented Feb 11, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm commented Feb 12, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pvary Feb 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pvary commented Feb 17, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Comments

swapna267 Feb 11, 2026 •

edited

Loading

pvary Feb 17, 2026 •

edited

Loading