{"record":{"id":"1486c75990107b1e","repo":"apache/iceberg","slug":"illegal-table-name","errorCode":null,"errorMessage":"Illegal table name:","messagePattern":"Illegal table name:","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java","lineNumber":166,"sourceCode":"    String[] namespace = new String[baseNamespace.levels().length + 1];\n    System.arraycopy(baseNamespace.levels(), 0, namespace, 0, baseNamespace.levels().length);\n    namespace[baseNamespace.levels().length] = newLevel;\n    return Namespace.of(namespace);\n  }\n\n  TableIdentifier toIdentifier(ObjectPath path) {\n    String objectName = path.getObjectName();\n    List<String> tableName = Splitter.on('$').splitToList(objectName);\n\n    if (tableName.size() == 1) {\n      return TableIdentifier.of(\n          appendLevel(baseNamespace, path.getDatabaseName()), path.getObjectName());\n    } else if (tableName.size() == 2 && MetadataTableType.from(tableName.get(1)) != null) {\n      return TableIdentifier.of(\n          appendLevel(appendLevel(baseNamespace, path.getDatabaseName()), tableName.get(0)),\n          tableName.get(1));\n    } else {\n      throw new IllegalArgumentException(\"Illegal table name:\" + objectName);\n    }\n  }\n\n  @Override\n  public List<String> listDatabases() throws CatalogException {\n    if (asNamespaceCatalog == null) {\n      return Collections.singletonList(getDefaultDatabase());\n    }\n\n    return asNamespaceCatalog.listNamespaces(baseNamespace).stream()\n        .map(n -> n.level(n.levels().length - 1))\n        .collect(Collectors.toList());\n  }\n\n  @Override\n  public CatalogDatabase getDatabase(String databaseName)\n      throws DatabaseNotExistException, CatalogException {\n    if (asNamespaceCatalog == null) {","sourceCodeStart":148,"sourceCodeEnd":184,"githubUrl":"https://github.com/apache/iceberg/blob/86d9c8fc543e7c56c9f624eb725f76c9baff9570/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/FlinkCatalog.java#L148-L184","documentation":"FlinkCatalog.toIdentifier converts a Flink ObjectPath/TablePath into an Iceberg TableIdentifier. When the table name has more parts than the catalog's namespace depth can absorb (e.g. a 3-level name under a 1-level base namespace, and it is not a metadata table), it throws IllegalArgumentException(\"Illegal table name:\").","triggerScenarios":"Calling table/tableExists/dropTable/renameTable or creating a table loader with an ObjectPath whose name contains extra dot-separated levels (size 3+ not a metadata table) or other shapes that don't fit baseNamespace + database + object.","commonSituations":"Flink catalogs configured with a base namespace where users reference nested Iceberg namespaces (a.b.table) in SQL; using Flink metadata-table syntax with a name that resolves to too many levels; copy-pasted fully qualified names from other engines.","solutions":["Reference the table with the correct number of levels for the catalog: use flinkCatalog name + database + table only, e.g. SELECT * FROM iceberg_db.`table` instead of iceberg_db.`ns.table.sub`.","Set the catalog's base-namespace property so the extra level maps onto a real Iceberg namespace (base-namespace = ns makes a.b.table valid).","Verify whether the second element is a valid metadata table type; only 2-level names ending in a metadata table keyword are accepted.","Query which databases exist via SHOW DATABASES to see the supported namespace layout."],"exampleFix":"// before: 3-level reference under flat catalog\nTableIdentifier id = flinkCatalog.toIdentifier(new ObjectPath(new ObjectPath(\"db\"), \"a.b.table\"));\n// after: configure base-namespace or use 2-level path\nTableIdentifier id = flinkCatalog.toIdentifier(new ObjectPath(new ObjectPath(\"db\"), \"a.table\"));","handlingStrategy":"validation","validationCode":"String[] parts = objectName.split(\"\\\\.\");\nif (parts.length > 2) throw new IllegalArgumentException(\"Too many levels for Flink Iceberg catalog: \" + objectName);","typeGuard":null,"tryCatchPattern":"try {\n  Table table = flinkCatalog.getTable(objectPath);\n} catch (IllegalArgumentException e) {\n  // fix identifier levels or base-namespace config\n}","preventionTips":["Use one database level + one table level in Flink SQL against Iceberg catalogs.","Set catalog base-namespace when you need an extra namespace level.","Check metadata-table syntax stays at exactly two levels."],"tags":["flink","catalog","naming"],"backgroundTag":"invalid-identifier-format","analyzedSha":"86d9c8fc543e7c56c9f624eb725f76c9baff9570","analyzedAt":"2026-09-12T00:46:39.097Z","contentChangedAt":"2026-09-12T00:46:39.097Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}