support create table like in flink catalog #12199
Conversation
| FlinkCreateTableOptions.CATALOG_TABLE.key(), tablePath.getObjectName()); | ||
| catalogAndTableProps.put("connector", FlinkDynamicTableFactory.FACTORY_IDENTIFIER); | ||
| catalogAndTableProps.putAll(table.properties()); | ||
| return toCatalogTableWithProps(table, catalogAndTableProps); |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
Just thinking out loud:
- Maybe the code would be easier to read if we send only the catalogProps to the
toCatalogTableWithPropsand 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?
| @@ -188,6 +188,23 @@ public void testCreateTableLike() throws TableNotExistException { | |||
| .isEqualTo(TableSchema.builder().field("id", DataTypes.BIGINT()).build()); | |||
| } | |||
|
|
|||
| @TestTemplate | |||
| public void testCreateTableLikeInFlinkCatalog() throws TableNotExistException { | |||
| sql("CREATE TABLE tl(id BIGINT)"); | |||
| @@ -384,13 +395,6 @@ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignor | |||
| @Override | |||
| public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) | |||
| throws CatalogException, TableAlreadyExistException { | |||
| if (Objects.equals( | |||
There was a problem hiding this comment.
Do we have a way to check if we can get the catalog properties in another way if they are not directly set by the table options?
| @@ -188,6 +188,23 @@ public void testCreateTableLike() throws TableNotExistException { | |||
| .isEqualTo(TableSchema.builder().field("id", DataTypes.BIGINT()).build()); | |||
| } | |||
|
|
|||
| @TestTemplate | |||
| public void testCreateTableLikeInFlinkCatalog() throws TableNotExistException { | |||
There was a problem hiding this comment.
Is there a simple way to check whether the create table like command is working even when the source table is not using the default catalog.
I'm a bit concerned that this test would pass if we simply use the default catalog for the new table instead copying the values from the source table.
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.
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 .