support create table like in flink catalog #12199

swapna267 · 2025-02-07T22:46:25Z

Creation of dynamic Iceberg table in Flink Catalog using the underlying physical Iceberg table using LIKE clause.

Currently (without the changes in PR), create table in flink catalog works by configuring flink connector as described in,
flink-connector

But that needs user to provide the schema for the table. A way to tackle that is to do create table LIKE using below DDL.

CREATE TABLE table_wm (
      eventTS AS CAST(t1 AS TIMESTAMP(3)),
      WATERMARK FOR eventTS AS SOURCE_WATERMARK()
) WITH (
  'connector'='iceberg',
  'catalog-name'='iceberg_catalog',
  'catalog-database'='testdb',
  'catalog-table'='t'
) LIKE iceberg_catalog.testdb.t;

Options like connector, catalog-name, catalog-database, catalog-table need to be duplicated as Iceberg FlinkCatalog doesn't return any catalog related properties during getTable. This PR addresses the issue by including these properties when getTable is called , which will be used by Flink when creating table in Flink Catalog.

Previous discussion related to feature is in PR, #12116 .

pvary · 2025-02-08T07:11:40Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

+        FlinkCreateTableOptions.CATALOG_TABLE.key(), tablePath.getObjectName());
+    catalogAndTableProps.put("connector", FlinkDynamicTableFactory.FACTORY_IDENTIFIER);
+    catalogAndTableProps.putAll(table.properties());
+    return toCatalogTableWithProps(table, catalogAndTableProps);


Could you help me understand why is the table properties needed to be added here?
We also send the table as a parameter. Wouldn't it be enough?

Just thinking out loud:

Maybe the code would be easier to read if we send only the catalogProps to the toCatalogTableWithProps and create a merged map when calling the Flink method

This is somewhat suboptimal as we create an extra map

Even if we decide to follow your approach, the parameter name of the method should reflect that at the declaration of toCatalogTableWithProps, and maybe some comments or javadoc should be nice there for future generations 😉

WDYT?

Yeah merged it here , as merging later on needs an extra map.

Renamed the method to toCatalogTableWithCustomProps and modified parameter names. Hope it's more readable now.

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java

pvary · 2025-02-19T10:02:08Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

+      } else if (!("connector".equalsIgnoreCase(entry.getKey())
+          || FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY.equalsIgnoreCase(entry.getKey()))) {


It took me some time to decipher this 😄
Could we please break up the negation so it doesn't span multiple lines?
Maybe:

} else if (!"connector".equalsIgnoreCase(entry.getKey()) && !FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY.equalsIgnoreCase(entry.getKey())) {

Or it is just me?
Also maybe a comment should be nice

pvary · 2025-02-20T09:12:18Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

+    String srcCatalogProps =
+        FlinkCreateTableOptions.toJson(
+            getName(), tablePath.getDatabaseName(), tablePath.getObjectName(), catalogProps);
+
+    ImmutableMap.Builder<String, String> mergedProps = ImmutableMap.builder();
+    mergedProps.put("connector", FlinkDynamicTableFactory.FACTORY_IDENTIFIER);
+    mergedProps.put(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY, srcCatalogProps);
+    mergedProps.putAll(table.properties());


Could we please add a comment here, that we store the catalog properties in the merged property list to work around the Flink API limitations?

pvary · 2025-02-20T09:16:13Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkDynamicTableFactory.java

-    tableProps.forEach(flinkConf::setString);

-    String catalogName = flinkConf.getString(CATALOG_NAME);
+    Map<String, String> mergedProps = mergeSrcCatalogProps(tableProps);


Either here, or in the javadoc for the mergeSrcCatalogProps please describe what we are doing, and why

pvary

I generally like this approach.
Please add some comments to the code for the future developers.
Otherwise looks good to me.

swapna267 · 2025-02-21T18:11:13Z

Added comments to the code.

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

stevenzwu · 2025-03-03T18:18:35Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java

+          .noDefaultValue()
+          .withDescription("Properties for the underlying catalog for iceberg table.");
+
+  public static final String SRC_CATALOG_PROPS_KEY = "src-catalog";


also does Flink create table ... like ... work for Hive tables as source? I assume no. otherwise, we can refer to how that is implemented.

It feels tacky to use special property key to carry over catalog props and source table identifier. Does it make sense to modify Flink API to support this more elegantly?

Right this is not supported in hive,
https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/hive-compatibility/hive-dialect/create/

Supported in Kafka, but yeah there is no concept of catalog , mostly flat properties of cluster for kafka connector.
Sure, can look into Flink API changes for long term, but as all connectors don't have concept of catalog or hierarchy like Iceberg, not sure how that works out.

yeah, Kafka doesn't have catalog concept. we can take this discussion separately. it won't be a blocker for this PR

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java

stevenzwu

waiting for CI to complete

stevenzwu · 2025-03-05T21:24:52Z

thanks @swapna267 for the contribution and @pvary for the review

swapna267 · 2025-03-06T00:58:16Z

Thanks @stevenzwu and @pvary for the review.

…1.18 and v1.19 backports #12199

support create table like in flink catalog

c193d17

github-actions bot added the flink label Feb 7, 2025

pvary reviewed Feb 8, 2025

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java Outdated Show resolved Hide resolved

pvary reviewed Feb 8, 2025

View reviewed changes

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java Show resolved Hide resolved

pvary reviewed Feb 8, 2025

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java Show resolved Hide resolved

use json value for source catalog properties

d4a464d

pvary reviewed Feb 19, 2025

View reviewed changes

fix checkstyle errors

44d45af

pvary reviewed Feb 20, 2025

View reviewed changes

pvary approved these changes Feb 20, 2025

View reviewed changes

add comments

b4d7ffe

stevenzwu reviewed Feb 26, 2025

View reviewed changes

swapna267 added 2 commits March 2, 2025 10:20

refactor and add checks

8f90cd5

fix tests

bf55619

stevenzwu reviewed Mar 3, 2025

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkCatalogTable.java Outdated Show resolved Hide resolved

stevenzwu reviewed Mar 3, 2025

View reviewed changes

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java Show resolved Hide resolved

stevenzwu reviewed Mar 3, 2025

View reviewed changes

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java Outdated Show resolved Hide resolved

stevenzwu reviewed Mar 3, 2025

View reviewed changes

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java Outdated Show resolved Hide resolved

stevenzwu reviewed Mar 3, 2025

View reviewed changes

fix testcase

6745fef

stevenzwu reviewed Mar 4, 2025

View reviewed changes

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java Outdated Show resolved Hide resolved

move location key and mark class as internal

6beeb2b

stevenzwu approved these changes Mar 5, 2025

View reviewed changes

stevenzwu merged commit ffe9ad5 into apache:main Mar 5, 2025
20 checks passed

swapna267 mentioned this pull request Mar 25, 2025

Flink: Support create table like and source watermark for flink sql to 1.18,1.19 #12643

Closed

swapna267 mentioned this pull request Mar 28, 2025

Flink: backport support create table like in flink catalog #12679

Merged

pvary pushed a commit that referenced this pull request Mar 30, 2025

Flink: Backport support create table like in flink catalog to Flink v…

d945cde

…1.18 and v1.19 backports #12199

		} else if (!("connector".equalsIgnoreCase(entry.getKey())
		\|\| FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY.equalsIgnoreCase(entry.getKey()))) {

support create table like in flink catalog #12199

support create table like in flink catalog #12199

Uh oh!

Conversation

swapna267 commented Feb 7, 2025

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!

Uh oh!

Uh oh!

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 left a comment

Choose a reason for hiding this comment

Uh oh!

swapna267 commented Feb 21, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

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!

Uh oh!

stevenzwu left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

stevenzwu commented Mar 5, 2025

Uh oh!

swapna267 commented Mar 6, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants