{"record":{"id":"564e2ff0815082b7","repo":"apache/iceberg","slug":"cannot-create-the-table-with-connector-iceberg","errorCode":null,"errorMessage":"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.","messagePattern":"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\\.","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java","lineNumber":420,"sourceCode":"          toIdentifier(new ObjectPath(tablePath.getDatabaseName(), newTableName)));\n    } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {\n      if (!ignoreIfNotExists) {\n        throw new TableNotExistException(getName(), tablePath, e);\n      }\n    } catch (AlreadyExistsException e) {\n      throw new TableAlreadyExistException(getName(), tablePath, e);\n    }\n  }\n\n  @Override\n  public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists)\n      throws CatalogException, TableAlreadyExistException {\n    // Creating Iceberg table using connector is allowed only when table is created using LIKE\n    if (Objects.equals(\n            table.getOptions().get(FlinkCreateTableOptions.CONNECTOR_PROPS_KEY),\n            FlinkDynamicTableFactory.FACTORY_IDENTIFIER)\n        && table.getOptions().get(FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY) == null) {\n      throw new IllegalArgumentException(\n          \"Cannot create the table with 'connector'='iceberg' table property in \"\n              + \"an iceberg catalog, Please create table with 'connector'='iceberg' property in a non-iceberg catalog or \"\n              + \"create table without 'connector'='iceberg' related properties in an iceberg table.\");\n    }\n\n    Preconditions.checkArgument(\n        table instanceof ResolvedCatalogTable,\n        \"Expected a ResolvedCatalogTable but got: %s. \"\n            + \"Iceberg Flink catalog only supports resolved catalog tables \"\n            + \"(Materialized tables and other table kinds are not supported).\",\n        table == null ? \"null\" : table.getClass().getName());\n    createIcebergTable(tablePath, (ResolvedCatalogTable) table, ignoreIfExists);\n  }\n\n  void createIcebergTable(ObjectPath tablePath, ResolvedCatalogTable table, boolean ignoreIfExists)\n      throws CatalogException, TableAlreadyExistException {\n    validateFlinkTable(table);\n","sourceCodeStart":402,"sourceCodeEnd":438,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java#L402-L438","documentation":"FlinkCatalog.createTable rejects DDL that arrives with the Flink 'connector'='iceberg' option when the table is being created directly inside an Iceberg catalog. Iceberg tables already live in the Iceberg catalog, so a connector option is only meaningful when bridging from a non-Iceberg catalog (e.g. a generic/Hive path via CREATE TABLE ... LIKE). If the connector key is present and there is no source-catalog marker, the request is contradictory and fails fast with IllegalArgumentException.","triggerScenarios":"Calling CREATE TABLE with OPTIONS('connector'='iceberg') against a catalog resolved as an Iceberg catalog (FlinkCatalog), without the SRC_CATALOG_PROPS_KEY option that the LIKE-based bridge path sets. Any engine path that forwards FlinkDynamicTableFactory.FACTORY_IDENTIFIER in table options to createTable.","commonSituations":"Following generic Flink connector docs ('connector'='iceberg' in WITH/OPTIONS) while the catalog itself is already an Iceberg catalog; mixing Hive catalog syntax with Iceberg catalog DDL; copying examples meant for CREATE TABLE LIKE workflows.","solutions":["Remove the 'connector'='iceberg' option (and related connector properties) when the catalog is already an Iceberg catalog — CREATE TABLE foo (...) suffices.","If bridging is intended, create the table in a non-Iceberg catalog with 'connector'='iceberg', or use CREATE TABLE ... LIKE so the SRC_CATALOG_PROPS_KEY marker is set.","Set the source-catalog properties option (FlinkCreateTableOptions.SRC_CATALOG_PROPS_KEY) if this table originates from another catalog's LIKE statement."],"exampleFix":"// before\nCREATE TABLE t WITH ('connector'='iceberg', 'path'='hdfs://...') AS SELECT ...\n// after\nCREATE TABLE t AS SELECT ...  -- when using an Iceberg catalog directly","handlingStrategy":"validation","validationCode":"boolean isIcebergCatalog = catalog instanceof org.apache.iceberg.flink.FlinkCatalog;\nboolean setsConnector = table.getOptions().get(\"connector\").equals(\"iceberg\");\nif (isIcebergCatalog && setsConnector) { throw new IllegalStateException(\"Do not pass connector=iceberg to an Iceberg catalog\"); }","typeGuard":"boolean isValidIcebergDdl(CatalogTable t) {\n  return !(\"iceberg\".equals(t.getOptions().get(\"connector\")));\n}","tryCatchPattern":null,"preventionTips":["Never put 'connector'='iceberg' in WITH/OPTIONS when the catalog is already an Iceberg catalog","Reserve connector=iceberg options for generic (non-Iceberg) catalogs or CREATE TABLE LIKE","Centralize table-creation helpers so connector options are set in one place"],"tags":["flink","iceberg-catalog","ddl","connector-options"],"backgroundTag":"conflicting-config-options","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}