support create table like in flink catalog

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

@@ -97,17 +97,20 @@ public class FlinkCatalog extends AbstractCatalog {
   private final Namespace baseNamespace;
   private final SupportsNamespaces asNamespaceCatalog;
   private final Closeable closeable;
+  private final Map<String, String> catalogProps;
   private final boolean cacheEnabled;
 
   public FlinkCatalog(
       String catalogName,
       String defaultDatabase,
       Namespace baseNamespace,
       CatalogLoader catalogLoader,
+      Map<String, String> catalogProps,
       boolean cacheEnabled,
       long cacheExpirationIntervalMs) {
     super(catalogName, defaultDatabase);
     this.catalogLoader = catalogLoader;
+    this.catalogProps = catalogProps;
     this.baseNamespace = baseNamespace;
     this.cacheEnabled = cacheEnabled;
 
@@ -332,7 +335,15 @@ public List<String> listTables(String databaseName)
   public CatalogTable getTable(ObjectPath tablePath)
       throws TableNotExistException, CatalogException {
     Table table = loadIcebergTable(tablePath);
-    return toCatalogTable(table);
+    Map<String, String> catalogAndTableProps = Maps.newHashMap(catalogProps);
+    catalogAndTableProps.put(FlinkCreateTableOptions.CATALOG_NAME.key(), getName());
+    catalogAndTableProps.put(
+        FlinkCreateTableOptions.CATALOG_DATABASE.key(), tablePath.getDatabaseName());
+    catalogAndTableProps.put(
+        FlinkCreateTableOptions.CATALOG_TABLE.key(), tablePath.getObjectName());
+    catalogAndTableProps.put("connector", FlinkDynamicTableFactory.FACTORY_IDENTIFIER);
+    catalogAndTableProps.putAll(table.properties());
+    return toCatalogTableWithProps(table, catalogAndTableProps);
   }
 
   private Table loadIcebergTable(ObjectPath tablePath) throws TableNotExistException {
@@ -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(
-        table.getOptions().get("connector"), FlinkDynamicTableFactory.FACTORY_IDENTIFIER)) {
-      throw new IllegalArgumentException(
-          "Cannot create the table with 'connector'='iceberg' table property in "
-              + "an iceberg catalog, Please create table with 'connector'='iceberg' property in a non-iceberg catalog or "
-              + "create table without 'connector'='iceberg' related properties in an iceberg table.");
-    }
     Preconditions.checkArgument(table instanceof ResolvedCatalogTable, "table should be resolved");
     createIcebergTable(tablePath, (ResolvedCatalogTable) table, ignoreIfExists);
   }
@@ -625,7 +629,7 @@ private static List<String> toPartitionKeys(PartitionSpec spec, Schema icebergSc
     return partitionKeysBuilder.build();
   }
 
-  static CatalogTable toCatalogTable(Table table) {
+  static CatalogTable toCatalogTableWithProps(Table table, Map<String, String> props) {
     TableSchema schema = FlinkSchemaUtil.toSchema(table.schema());
     List<String> partitionKeys = toPartitionKeys(table.spec(), table.schema());
 
@@ -634,7 +638,11 @@ static CatalogTable toCatalogTable(Table table) {
     // CatalogTableImpl to copy a new catalog table.
     // Let's re-loading table from Iceberg catalog when creating source/sink operators.
     // Iceberg does not have Table comment, so pass a null (Default comment value in Flink).
-    return new CatalogTableImpl(schema, partitionKeys, table.properties(), null);
+    return new CatalogTableImpl(schema, partitionKeys, props, null);
+  }
+
+  static CatalogTable toCatalogTable(Table table) {
+    return toCatalogTableWithProps(table, table.properties());
   }
 
   @Override

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

@@ -168,6 +168,7 @@ protected Catalog createCatalog(
         defaultDatabase,
         baseNamespace,
         catalogLoader,
+        properties,
         cacheEnabled,
         cacheExpirationIntervalMs);
   }

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

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+public class FlinkCreateTableOptions {
+
+  private FlinkCreateTableOptions() {}
+
+  public static final ConfigOption<String> CATALOG_NAME =
+      ConfigOptions.key("catalog-name")
+          .stringType()
+          .noDefaultValue()
+          .withDescription("Catalog name");
+
+  public static final ConfigOption<String> CATALOG_TYPE =
+      ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
+          .stringType()
+          .noDefaultValue()
+          .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
+
+  public static final ConfigOption<String> CATALOG_DATABASE =
+      ConfigOptions.key("catalog-database")
+          .stringType()
+          .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
+          .withDescription("Database name managed in the iceberg catalog.");
+
+  public static final ConfigOption<String> CATALOG_TABLE =
+      ConfigOptions.key("catalog-table")
+          .stringType()
+          .noDefaultValue()
+          .withDescription("Table name managed in the underlying iceberg catalog and database.");
+}

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

@@ -21,7 +21,6 @@
 import java.util.Map;
 import java.util.Set;
 import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.catalog.CatalogDatabaseImpl;
@@ -45,31 +44,6 @@
 public class FlinkDynamicTableFactory
     implements DynamicTableSinkFactory, DynamicTableSourceFactory {
   static final String FACTORY_IDENTIFIER = "iceberg";
-
-  private static final ConfigOption<String> CATALOG_NAME =
-      ConfigOptions.key("catalog-name")
-          .stringType()
-          .noDefaultValue()
-          .withDescription("Catalog name");
-
-  private static final ConfigOption<String> CATALOG_TYPE =
-      ConfigOptions.key(FlinkCatalogFactory.ICEBERG_CATALOG_TYPE)
-          .stringType()
-          .noDefaultValue()
-          .withDescription("Catalog type, the optional types are: custom, hadoop, hive.");
-
-  private static final ConfigOption<String> CATALOG_DATABASE =
-      ConfigOptions.key("catalog-database")
-          .stringType()
-          .defaultValue(FlinkCatalogFactory.DEFAULT_DATABASE_NAME)
-          .withDescription("Database name managed in the iceberg catalog.");
-
-  private static final ConfigOption<String> CATALOG_TABLE =
-      ConfigOptions.key("catalog-table")
-          .stringType()
-          .noDefaultValue()
-          .withDescription("Table name managed in the underlying iceberg catalog and database.");
-
   private final FlinkCatalog catalog;
 
   public FlinkDynamicTableFactory() {
@@ -127,16 +101,16 @@ public DynamicTableSink createDynamicTableSink(Context context) {
   @Override
   public Set<ConfigOption<?>> requiredOptions() {
     Set<ConfigOption<?>> options = Sets.newHashSet();
-    options.add(CATALOG_TYPE);
-    options.add(CATALOG_NAME);
+    options.add(FlinkCreateTableOptions.CATALOG_TYPE);
+    options.add(FlinkCreateTableOptions.CATALOG_NAME);
     return options;
   }
 
   @Override
   public Set<ConfigOption<?>> optionalOptions() {
     Set<ConfigOption<?>> options = Sets.newHashSet();
-    options.add(CATALOG_DATABASE);
-    options.add(CATALOG_TABLE);
+    options.add(FlinkCreateTableOptions.CATALOG_DATABASE);
+    options.add(FlinkCreateTableOptions.CATALOG_TABLE);
     return options;
   }
 
@@ -153,14 +127,17 @@ private static TableLoader createTableLoader(
     Configuration flinkConf = new Configuration();
     tableProps.forEach(flinkConf::setString);
 
-    String catalogName = flinkConf.getString(CATALOG_NAME);
+    String catalogName = flinkConf.getString(FlinkCreateTableOptions.CATALOG_NAME);
     Preconditions.checkNotNull(
-        catalogName, "Table property '%s' cannot be null", CATALOG_NAME.key());
+        catalogName,
+        "Table property '%s' cannot be null",
+        FlinkCreateTableOptions.CATALOG_NAME.key());
 
-    String catalogDatabase = flinkConf.getString(CATALOG_DATABASE, databaseName);
+    String catalogDatabase =
+        flinkConf.getString(FlinkCreateTableOptions.CATALOG_DATABASE, databaseName);
     Preconditions.checkNotNull(catalogDatabase, "The iceberg database name cannot be null");
 
-    String catalogTable = flinkConf.getString(CATALOG_TABLE, tableName);
+    String catalogTable = flinkConf.getString(FlinkCreateTableOptions.CATALOG_TABLE, tableName);
     Preconditions.checkNotNull(catalogTable, "The iceberg table name cannot be null");
 
     org.apache.hadoop.conf.Configuration hadoopConf = FlinkCatalogFactory.clusterHadoopConf();

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

@@ -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)");
+
+    sql("CREATE TABLE `default_catalog`.`default_database`.tl2 LIKE tl");
+
+    CatalogTable catalogTable = catalogTable("default_catalog", "default_database", "tl2");
+    assertThat(catalogTable.getSchema())
+        .isEqualTo(TableSchema.builder().field("id", DataTypes.BIGINT()).build());
+
+    Map<String, String> options = catalogTable.getOptions();
+    assertThat(options.entrySet().containsAll(config.entrySet())).isTrue();
+    assertThat(options.get(FlinkCreateTableOptions.CATALOG_NAME.key())).isEqualTo(catalogName);
+    assertThat(options.get(FlinkCreateTableOptions.CATALOG_DATABASE.key())).isEqualTo(DATABASE);
+    assertThat(options.get(FlinkCreateTableOptions.CATALOG_TABLE.key())).isEqualTo("tl");
+  }
+
   @TestTemplate
   public void testCreateTableLocation() {
     assumeThat(isHadoopCatalog)
@@ -660,10 +677,12 @@ private Table table(String name) {
   }
 
   private CatalogTable catalogTable(String name) throws TableNotExistException {
+    return catalogTable(getTableEnv().getCurrentCatalog(), DATABASE, name);
+  }
+
+  private CatalogTable catalogTable(String catalog, String database, String table)
+      throws TableNotExistException {
     return (CatalogTable)
-        getTableEnv()
-            .getCatalog(getTableEnv().getCurrentCatalog())
-            .get()
-            .getTable(new ObjectPath(DATABASE, name));
+        getTableEnv().getCatalog(catalog).get().getTable(new ObjectPath(database, table));
   }
 }

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

@@ -256,43 +256,6 @@ public void testCatalogDatabaseConflictWithFlinkDatabase() {
         .hasMessageStartingWith("Could not execute CreateTable in path");
   }
 
-  @TestTemplate
-  public void testConnectorTableInIcebergCatalog() {
-    // Create the catalog properties
-    Map<String, String> catalogProps = Maps.newHashMap();
-    catalogProps.put("type", "iceberg");
-    if (isHiveCatalog()) {
-      catalogProps.put("catalog-type", "hive");
-      catalogProps.put(CatalogProperties.URI, CatalogTestBase.getURI(hiveConf));
-    } else {
-      catalogProps.put("catalog-type", "hadoop");
-    }
-    catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, createWarehouse());
-
-    // Create the table properties
-    Map<String, String> tableProps = createTableProps();
-
-    // Create a connector table in an iceberg catalog.
-    sql("CREATE CATALOG `test_catalog` WITH %s", toWithClause(catalogProps));
-    try {
-      assertThatThrownBy(
-              () ->
-                  sql(
-                      "CREATE TABLE `test_catalog`.`%s`.`%s` (id BIGINT, data STRING) WITH %s",
-                      FlinkCatalogFactory.DEFAULT_DATABASE_NAME,
-                      TABLE_NAME,
-                      toWithClause(tableProps)))
-          .cause()
-          .isInstanceOf(IllegalArgumentException.class)
-          .hasMessage(
-              "Cannot create the table with 'connector'='iceberg' table property in an iceberg catalog, "
-                  + "Please create table with 'connector'='iceberg' property in a non-iceberg catalog or "
-                  + "create table without 'connector'='iceberg' related properties in an iceberg table.");
-    } finally {
-      sql("DROP CATALOG IF EXISTS `test_catalog`");
-    }
-  }
-
   private Map<String, String> createTableProps() {
     Map<String, String> tableProps = Maps.newHashMap(properties);
     tableProps.put("catalog-name", catalogName);