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

  }
+
+  @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

+    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.

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

+
+  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.

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

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

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

+            .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

+    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:47:43Z

+
+  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?

Don't see any tests also covering that. No impact on users, unless they are relying specifically on NPE here to do something specific.
checkNotNull is being used at multiple places in this class though. To keep it consistent, we need to apply at all places though.

Please change this to:

Preconditions.checkArgument(catalogName, "Invalid catalog name: null. Set %s table property.");

Sure. Cleaned up other nonnull checks and removed following 2 as they will never happen.

Preconditions.checkNotNull(catalog, "Flink catalog cannot be null"); -> createtableloader is only called when catalog != null.

Preconditions.checkNotNull(catalogTable, "The iceberg table name cannot be null"); -> Table name can never be null as create table statement will not compile without that.

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

Nice stuff @swapna267!

Could you please update the documentation too?

swapna267 · 2026-02-19T02:10:23Z

Thanks @pvary . Working on adding some usable dynamic record generators which can use VARIANT column type in Flink SQL to generate the dynamic records.Should have the PR by end of this week. Would like to add the documentation post that PR review.

pvary · 2026-02-19T15:48:21Z

-              .orElseGet(ImmutableList::of);
+    return (DataStreamSinkProvider)
+        (providerContext, dataStream) -> {
+          if (catalogLoader != null && dynamicRecordGeneratorImpl != null) {


Do we need a check somewhere that these things are set together?

Added a check by using useDynamicSink field to determine.

pvary · 2026-02-19T15:57:47Z

-          (providerContext, dataStream) -> {
            if (Boolean.TRUE.equals(
                readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_V2_SINK))) {
              return IcebergSink.forRowData(dataStream)


Could we just create a method of all of the sink creation, like

return createDynamicIcebergSink(dataStream); return legacySink(dataStream, resolvedSchema, equalityColumns); return icebergSink(dataStream, resolvedSchema, equalityColumns);

mxm

LGTM. Great work @swapna267! Thanks for reviewing, @pvary!

pvary · 2026-02-23T12:54:29Z

Merged to main.
Thanks @swapna267 for the PR, and @mxm and @Guosmilesmile for the review!

pvary · 2026-02-23T12:55:22Z

@swapna267: Please backport with something like this:

g d HEAD^ HEAD spark/v4.1 |sed "s/v4.1/v4.0/g">/tmp/patch;g apply -3 -p1 /tmp/patch

swapna267 · 2026-02-23T16:39:59Z

Thanks @pvary for review and merging. Will do the backport.

Backports #15279

Backports apache#15279

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

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

pvary reviewed Feb 17, 2026

View reviewed changes

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

pvary reviewed Feb 17, 2026

View reviewed changes

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

pvary reviewed Feb 17, 2026

View reviewed changes

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

swapna267 added 2 commits February 18, 2026 18:12

checkargument instead of checknotnull

e59ec5b

spotless fix

fe4e947

pvary reviewed Feb 19, 2026

View reviewed changes

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

pvary reviewed Feb 19, 2026

View reviewed changes

swapna267 added 3 commits February 19, 2026 10:22

pr feedback

d1a90ac

refactor icebergtablesink

f127560

fix checkstyle errors

0a5577e

pvary approved these changes Feb 23, 2026

View reviewed changes

mxm approved these changes Feb 23, 2026

View reviewed changes

pvary merged commit 86defce into apache:main Feb 23, 2026
14 checks passed

swapna267 mentioned this pull request Feb 25, 2026

Flink: Backport SQL support for dynamic iceberg sink #15444

Merged

pvary pushed a commit that referenced this pull request Feb 27, 2026

Flink: Backport SQL support for dynamic iceberg sink (#15444)

10de214

Backports #15279

RjLi13 pushed a commit to RjLi13/iceberg that referenced this pull request Mar 12, 2026

Flink: SQL support for dynamic iceberg sink (apache#15279)

d2ec808

RjLi13 pushed a commit to RjLi13/iceberg that referenced this pull request Mar 12, 2026

Flink: Backport SQL support for dynamic iceberg sink (apache#15444)

94d1366

Backports apache#15279

talatuyarer pushed a commit to talatuyarer/iceberg that referenced this pull request Apr 1, 2026

Flink: SQL support for dynamic iceberg sink (apache#15279)

1de04e4

talatuyarer pushed a commit to talatuyarer/iceberg that referenced this pull request Apr 1, 2026

Flink: Backport SQL support for dynamic iceberg sink (apache#15444)

53c4cdf

Backports apache#15279

Uh oh!

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!

Uh oh!

Uh oh!

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!

Choose a reason for hiding this comment

Uh oh!

swapna267 Feb 19, 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!

Uh oh!

pvary commented Feb 17, 2026

Uh oh!

swapna267 commented Feb 19, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pvary commented Feb 23, 2026

Uh oh!

pvary commented Feb 23, 2026

Uh oh!

swapna267 commented Feb 23, 2026

swapna267 Feb 11, 2026 •

edited

Loading

pvary Feb 17, 2026 •

edited

Loading

swapna267 Feb 19, 2026 •

edited

Loading