@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.");
        }

        createIcebergTable(tablePath, table, ignoreIfExists);
    }
Append a new level to the base namespace
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/FlinkCatalog.java